This repository contains code for a sample Spark job written in Python. Code processes GitHub events data and aggregates some information about them.

More about source data at: http://www.gharchive.org/

Structure of output data in Results section.

Run these commands:

(change directory 'F:/spark-task' for the directory where you cloned this repo on your machine)

$ docker run -d -p 8888:8888 -p 4040:4040 -p 4041:4041 -p 4042:4042 --name sparky -v F:/spark-task:/home/jovyan/work jupyter/all-spark-notebook:13b866ff67b9
$ cd /directory where you cloned this repo/
$ bash ./download_data.sh
$ docker exec -it sparky python3 ./work/test_aggregate.py
$ docker exec -i sparky /usr/local/spark/bin/spark-submit --master local[3] /home/jovyan/work/aggregate.py file:///home/jovyan/work/raw_data/*.json.gz file:///home/jovyan/work/output_data/

I used Jupyter's docker images to set up Spark. Their images create singe machine by default so no dedicated workers but it should be enough for this task.

You can run the container using command (change directory 'F:/spark-task' for the directory where you cloned this repo on your machine):

docker run -d -p 8888:8888 -p 4040:4040 -p 4041:4041 -p 4042:4042 --name sparky -v F:/spark-task:/home/jovyan/work jupyter/all-spark-notebook:13b866ff67b9

You can also choose another name for the container by changing --name parameter (if you do remember to use your name in the rest of the commands).

Container maps internal directory /home/jovyan/work to the directory on the host machine so Spark can access the files. In a regular production setup files would probably be stored in HDFS or S3.

Ports 8888, 4040-2 are mapped to the same ports on your machine. Port 8888 can be used to access jupyter notebook. Ports 4040 to 4042 are used by Spark Web UI which is created when Spark Context is created.

If you want to access jupyter you can run:

docker logs --tail 3 sparky

just after launching container to get the url with the token required by jupyter on launch.

When a job is running you can track stats and see details like DAG using Spark Web UI, server for that is only online when Spark Context is active so after job is run it might shut down. These contexts/servers take ports starting at 4040 and incrementing when given port is already in use that's why the container is mapping a few ports just in case you need too access ui for multiple contexts.

URL for web UI when it's active:

http://localhost:4040/jobs/

(or another port as described above)

Data can be downloaded using wget command.

You can use script in repo: download_data.sh to download files for the entire month (2018-01). Files will be saved in raw_data directory.

Example command:

$ cd /directory where you cloned this repo/
$ bash ./download_data.sh

Code is in Python. I opted to use DataFrame API since it's more similar to technologies I am familiar with.

Code could be rewritten using map/reduce concepts eg:

Iterate over records in files and map them with a function emitting key consisting of date, user id, user name, event type and value 1. There would be a few of those functions mapping for different aggregates.

Then use reducer function to sum those values.

Then pivot the data into a table and export it to an optimized format such as orc or parquet or avro which should be much faster to access and much more compact than json or csv.

Some basic tests are present. You can launch them using this command:

docker exec -it sparky python3 ./work/test_aggregate.py

These tests could use improvement as they don't cover many scenarios. There is only single row of each type in the test data.

To submit job to Spark in container you can use command like:

docker exec -i sparky /usr/local/spark/bin/spark-submit --master local[3] /home/jovyan/work/aggregate.py file:///home/jovyan/work/raw_data/2018-01-01-0.json.gz file:///home/jovyan/work/output_data/

Since this docker image does not provide a cluster we will use "local" as master. Specify number of available cores with the number in square brackets.

On a production-grade cluster we would specify more parameters that could eg limit resources for the job, pick resource manager and master location etc.

Then you provide path to the entry point/class ie script to run. (remember we defined docker volume mapping for /home/jovyan/work)

Then you provide file path for file(s) with the data. You can use wildcards in the filepath.

Last parameter is directory where we save the results.

Filepaths specify protocol for access since depending on environment configuration it could be eg S3 or HDFS storage. In this case we are just using file protocol.

To process all the downloaded data run:

docker exec -i sparky /usr/local/spark/bin/spark-submit --master local[3] /home/jovyan/work/aggregate.py file:///home/jovyan/work/raw_data/*.json.gz file:///home/jovyan/work/output_data/

Results are saved to two partitioned Parquet files.

Parquet format provides very good read performance and keeps the size of the files small.

Below there are some example rows from the final DataFrame to show the structure.

Data about users:

Data about repositories: