Part 1: The Data for This Workshop
In this workshop we will be using data from the USASpending.gov website. We will be looking at U.S. government contracts from 2014 through 2018. This period spans two different U.S. administrations, under two different presidents, from different political parties. The active members of these two parties have very different perspectives on governing and spending priorities. Are these differences reflected in the contracting being done? Do different agencies spend more? Are contract lengths different? Are the vendors for contracts different? Are the states where contracts are awarded different? Possibly, or it could be that the administration of contracts by career government employees does not change that much from administration to administration.
The USASpending site grew out of a government initiative on transparency as to how the government is spending money. Details regarding the application programming interface (API) for this data can be found here. That page includes a brief tutorial, and the Endpoints link on that page shows a list of the methods available to download data.
Some summary data can be downloaded directly in a JSON format and we will briefly cover an example at the end of this notebook. JSON has become the language of the web, so if you are downloading data from social media sites (e.g., Twitter, Yelp, etc.) or from other sources, JSON is a common format. Government and financial data often use an XML format, and scientific data usually has different formats.
For other larger data sets on the USASpending site, you can send an asynchronous request and your data is later available for download in a comma-separated value (CSV) format.
The APIs at the USASpending website will allow you to download data about contracts awarded, assistance (grants and loans) awarded, spending on those contracts and awards, federal accounts, and account balances.
Downloading the contracts data
For purposes of this seminar, we are going to start with data on U.S. government spending (federal) on contracts and who was being paid in each of the quarters mentioned above. Due to some quirks in the API (there is a 1GB limit that is silently imposed), we downloaded these by month, filtered the columns we wanted, combined the months by year, compressed the data, and staged the data files on the web so you can access them directly from your notebook (so we don't need to move large files over the Wi-Fi). Storing your data on S3 is relatively cheap - it costs pennies per GB per month.
Keep in mind that Spark can read compressed data files. As an example, the government spending data we will be using today compresses by a factor of 10 as a bzip2 file compared to the raw data, so the storage cost drops by a factor of 10.
For this workshop we filtered out some of the columns. The original contract data contained 250 columns and this was reduced to the 74 columns relevant to the questions we will ask of the data today (or other questions you may want to ask later). The actual downloading, filtering, compressing, etc. was done using a separate Jupyter notebook that does not require Spark (you can run that from Jupyter as part of Anaconda's free desktop distro). In that notebook we specify which columns of data we want, so you can specify different columns. If you want to download the data yourself or explore different columns in the USASpending data, see the "Tell Me More!" module in Canvas for this seminar.
Spinning Up a Cluster
To be able to calculate any cells in your notebook, you will need to be attached to a cluster. From the toolbar on the
left-hand side, click on the Compute icon. You won't have any clusters to start with, so click on the +Create Cluster button.
There are two fields you need to specify:
- You need to enter a name for your cluster
- There is a DataBricks Runtime (DBR) that is the default. From the drop-down list for the DBR, select version 9.0 or higher, but do not select the "ML" version. If you select a version lower than 9.0, the data entry widget you will be creating in the next cell will not appear (and that's not fun). The drop-down list is shown in the image below.
After entering a name for your cluster and specifying the DBR version, click the "Create Cluster" button. Initially you will see a spinning green circle next to the name of your cluster. That indicates that the cluster is starting (what's also referred to as "spinning up"). Once your cluster is up and running (it will have a solid green circle next to it), come back here to your notebook and from the drop-down list in the top-left corner that says "Detached", select your running cluster (listed with the solid green circle next to it). Your notebook will now be attached to your running cluster and can execute code cells, so let's get started!
Creating The Input Widget
When you run the following cell, Databricks will display a widget at the top of the notebook that prompts you for the URL for a data manifest. The URL is the location on the web where you can find the file, and the URL for the seminar is provided in Canvas. The data manifest lists the files you will need and tells the notebook where the files are located and some information that the notebook will use to verify that the files were retrieved without any problems.
To run the following cell. hover your mouse over the cell and some controls will appear in the upper right-hand corner of the cell as shown in the image below.
. Select the Run cell option.
The following cell is Step 1a
import pyspark.sql.functions as f
import requests
import re
import tarfile
import zipfile
import json
import hashlib
FORCE_DOWNLOAD = False
MANIFEST_URL = 'https://www.sjsu.edu/datascienceforall/seminars/spark-jupyter-manifest.json'
URL_HOST = ".s3-us-west-2.amazonaws.com/"
TEMP_DIR = "/ds4all/"
ZIPPED_DIR = "/ds4all/zipped/"
UNZIPPED_DIR = "/ds4all/unzipped/"
DATA_DIR = "/user/hive/warehouse/usa_transactions_table"
DB_NAME = "usa_transactions_table"
def cleanAll():
' Removes the database directory if it exists on DBFS'
spark.sql(f"DROP TABLE IF EXISTS {DB_NAME}")
try:
dbutils.fs.rm(DATA_DIR,recurse=True)
except Exception:
pass
def get_manifest():
''' Returns the dictionary that's the download manifest based on the URL
entered in the URL widget. If it's not a valid URL or returns a status code other than 200, an exception is raised. If the manifest is not valid JSON, or does not contain name:value pairs named id, data_list, and download_list, an exception is raised. '''
manifest_url = dbutils.widgets.get("manifest_url")
response = requests.get(manifest_url)
if response.status_code != 200:
raise Exception(f"The manifest URL {manifest_url} returned a status of " + str(response.status))
manifest = response.text
try:
manifest_dict = json.loads(manifest)
# The manifest should have a data_list element and a down_Load list element
if "data_list" in manifest_dict == False:
raise Exception("The manifest does not contain a data_list.")
if "download_list" in manifest_dict == False:
raise Exception("The manifest does not contain a download_list.")
return(manifest_dict)
except json.JSONDecodeError as err:
raise Exception("The manifest is not a valid JSON document.", err)
def check_data(manifest):
''' Function used to check if the data directory contains valid
copies of all of the files or directories in the download. The manifest dictionary is passed as a parameter and is expected to comtain a data_list containing the names of each file or directory expected in the data directory. If a file or directory is missing, this method returns False. If all of the files or directories exist, it returns True. '''try:
data_dir_list = dbutils.fs.ls(DATA_DIR)
if len(db_dir_list) == 0:
return(False)
file_list = manifest["file_list"]
existing_list = dbutils.fs.ls(DATA_DIR)
for file_name in file_list:
found == False
for info in existing_list:
if info.name == file_name:
found == True
break
if found == False:
return(False)
# looped through all of the required files and they are there
return(True)
except Exception:
# The directory does not exist or does not match the manifest
return(False)
def get_bucket_id(manifest):
''' The manifest is expected to contain a name:value pair named id
where the value is the bucket name on S3 where the files are staged. If the id is missing or is a blanks string, then an exception is raised, otherwise the bucket id is returned. '''try:
bucket = manifest['id'].strip()
if len(bucket) == 0:
raise Exception("The id provided in the manifest was an empty string, but should be the name of the bucket being downloaded from.")
else:
return(bucket)
except Exception as e:
raise Exception("An error occurred in retrieving the bucket id from the manifest", e)
def download_file(manifest_item, bucket_id):
''' Given a dictionary from the download list, download the file to the
temporary directory for downloading the file and check the MD5 sum to make sure it matches. If the MD5 sum does not match, an excetion is raised, otherwise it prints that the file was successfully downloaded. '''file_name = manifest_item["name"]
item_md5sum = manifest_item["md5"]
request_url = "https://" + bucket_id + URL_HOST + file_name
local_name = ZIPPED_DIR + file_name
print("requesting file from:", request_url)
r = requests.get(request_url, stream=True)
status_code = r.status_code
# If the status code is 200, then we successfully retrieved the file
if status_code != 200:
raise Exception(f"The {file_name} download failed. A status code of {str(status_code)} was returned from the URL:{request_url}.")
else: # write the file
with open(local_name, 'wb') as file:
for chunk in r.iter_content(chunk_size=4096):
file.write(chunk)
file.flush()
file.close()
#check if the hash of the file downloaded matches the md5 sum in the manifest
with open(local_name, 'rb') as data_file:
md5sum = hashlib.md5( data_file.read() ).hexdigest()
if md5sum.lower() != item_md5sum.lower():
raise Exception(f"The file {file_name} downloaded from Google Drive generated a MD5 sum of {md5sum} instead of the MD5 sum in the manifest ({item_md5sum}) so it may be corrupted and the processing was terminated.")
else:
print ("successfully downloaded:", file_name)
def process_file(manifest_item):
''' The file is now downloaded. If the file is zipped,
it first needs to be unziiped, and either way, moved to the DBFS data directory. '''local_name = ZIPPED_DIR + manifest_item["name"]
local_path = "file:" + local_name
is_zipped = manifest_item["zipped"] == "true" # This is either Ture or False
if is_zipped:
with tarfile.open(name=local_name,mode='r:bz2') as tar_ref:
tar_ref.extractall(path=UNZIPPED_DIR)
untar_info = dbutils.fs.ls("file:" + UNZIPPED_DIR)
# The zip file could contain a directory, a file, or more than 1 file,
# so we loop through the file list, moving all of them to DBFS
for info in untar_info:
destination = DATA_DIR + "/" + info.name
dbutils.fs.mv(info.path, destination, recurse=True)
dbutils.fs.rm(local_path)
else: # file was not zipped (or should remain zipped), so just move it
destination = DATA_DIR + "/" + manifest_item["name"]
dbutils.fs.mv(local_path, destination)
print ("processed:", local_name)
def load_data(manifest_list, bucket_id):
''' Loops through the files in the download list from the manifest and
downloads the file, verifies the MD5 sum is correct, unzips it if needed, and moves the file or folder that was in it to the data directory.'''# Create the empty temporary directories
try:
dbutils.fs.rm("file:" + TEMP_DIR,recurse=True)
except Exception:
pass
# Create the temporary local directory and sub-directories
dbutils.fs.mkdirs("file:" + TEMP_DIR)
dbutils.fs.mkdirs("file:" + ZIPPED_DIR)
dbutils.fs.mkdirs("file:" + UNZIPPED_DIR)
# Loop through the files to download
for item in manifest_list:
download_file(item, bucket_id)
process_file(item)
# Remove the temp directory used to unzip the files
dbutils.fs.rm("file:" + TEMP_DIR, recurse=True)
def createTable():
spark.sql(f"DROP TABLE IF EXISTS {DB_NAME}")
spark.sql(f"""
CREATE TABLE {DB_NAME}
USING PARQUET LOCATION '{DATA_DIR}'
""")
spark.sql(f"MSCK REPAIR TABLE {DB_NAME}")
# ******************************************* # Run the Actual Routine to Load the Data# This code uses the above defined functions# *******************************************if FORCE_DOWNLOAD == True:
cleanAll()
if ( spark.catalog._jcatalog.tableExists(DB_NAME) != True):
manifest_dict = get_manifest()
if check_data(manifest_dict) == False:
cleanAll()
bucket_id = get_bucket_id(manifest_dict)
download_list = manifest_dict["download_list"]
load_data(download_list, bucket_id)
else:
print("All of the required files exist in the data directory already, so the download was not processed.")
# The table definition is deleted when the cluster shuts down, but the files remain.
# The call to createTable will create the able if we just loaded the files or if they were already there
createTable()
else:
print("The table already exists. If you want to recreate it, set the FORCE_DOWNLOAD constant to true and re-run this cell.")
df_transactions = spark.sql("""
""")
df_transactions.show(2,truncate=False, vertical=True)
df_transactions.printSchema()
This Notebook
This notebook was created to help introduce students to the wonderful world of data science using Spark and Jupyter notebooks. The target audience are students who have no background in data science, programming, distributed computing, Python, or Apache Spark.
This notebook is designed to run on a Databricks Community account (our thanks to Databricks for making community accounts available!).
This notebook was created as part of the Data Science for All Seminar Series funded by NSF Grant #1829622
PI: Leslie Albert, Ph.D., Co-PIs: Scott Jensen, Ph.D. and Esperanza Huerta, Ph.D.
Copyright Scott Jensen, San Jose State University
This ds4all notebook by Scott Jensen,Ph.D. is licensed under a Creative Commons Attribution-ShareAlike 4.0 International License.