wanlipu / airflow-de

Project - Data Pipelines with Airflow

Project Context and Purpose

A music streaming company, Sparkify, has decided that it is time to introduce more automation and monitoring to their data warehouse ETL pipelines and come to the conclusion that the best tool to achieve this is Apache Airflow.

They have decided to bring you into the project and expect you to create high grade data pipelines that are dynamic and built from reusable tasks, can be monitored, and allow easy backfills. They have also noted that the data quality plays a big part when analyses are executed on top the data warehouse and want to run tests against their datasets after the ETL steps have been executed to catch any discrepancies in the datasets.

The source data resides in S3 and needs to be processed in Sparkify's data warehouse in Amazon Redshift. The source datasets consist of JSON logs that tell about user activity in the application and JSON metadata about the songs the users listen to.

Project Description

In this project, I will apply Apache Airflow and data pipelines to build an ETL pipeline. To complete the project, I will need to load data from S3, process the data into analytics tables using Apache Airflow, and load them back into S3.

Dataset

Song Dataset

The first dataset is a subset of real data from the Million Song Dataset. Each file is in JSON format and contains metadata about a song and the artist of that song. The files are partitioned by the first three letters of each song's track ID. For example, here are filepaths to two files in this dataset.

song_data/A/B/C/TRABCEI128F424C983.json
song_data/A/A/B/TRAABJL12903CDCF1A.json

And below is an example of what a single song file, TRAABJL12903CDCF1A.json, looks like.

{"num_songs": 1, "artist_id": "ARJIE2Y1187B994AB7", "artist_latitude": null, "artist_longitude": null, "artist_location": "", "artist_name": "Line Renaud", "song_id": "SOUPIRU12A6D4FA1E1", "title": "Der Kleine Dompfaff", "duration": 152.92036, "year": 0}

Log Dataset

The second dataset consists of log files in JSON format generated by this event simulator based on the songs in the dataset above. These simulate activity logs from a music streaming app based on specified configurations.

The log files in the dataset you'll be working with are partitioned by year and month. For example, here are filepaths to two files in this dataset

log_data/2018/11/2018-11-12-events.json
log_data/2018/11/2018-11-13-events.json

And below is an example of what part of a single log file, 2018-11-01-events.json, looks like.

{"artist":"Black Eyed Peas","auth":"Logged In","firstName":"Sylvie","gender":"F","itemInSession":0,"lastName":"Cruz","length":214.93506,"level":"free","location":"Washington-Arlington-Alexandria, DC-VA-MD-WV","method":"PUT","page":"NextSong","registration":1540266185796.0,"sessionId":9,"song":"Pump It","status":200,"ts":1541108520796,"userAgent":"\"Mozilla\/5.0 (Macintosh; Intel Mac OS X 10_9_4) AppleWebKit\/537.77.4 (KHTML, like Gecko) Version\/7.0.5 Safari\/537.77.4\"","userId":"10"}
{"artist":null,"auth":"Logged In","firstName":"Ryan","gender":"M","itemInSession":0,"lastName":"Smith","length":null,"level":"free","location":"San Jose-Sunnyvale-Santa Clara, CA","method":"GET","page":"Home","registration":1541016707796.0,"sessionId":169,"song":null,"status":200,"ts":1541109015796,"userAgent":"\"Mozilla\/5.0 (X11; Linux x86_64) AppleWebKit\/537.36 (KHTML, like Gecko) Ubuntu Chromium\/36.0.1985.125 Chrome\/36.0.1985.125 Safari\/537.36\"","userId":"26"}

Project Strucute

• README.md: instructions and documentation of the project
• ./images/airflow_dag.png: DAG visualization
• ./airflow/dags/udacity_dag.py: Directed Acyclic Graph definition with imports, tasks and task dependencies
• ./airflow/dags/create_tables.sql: SQL Table creations statements
• ./airflow/plugins/helpers/sql_queries.py: Insert SQL statements
• ./airflow/plugins/operators/stage_redshift.py: Operator that copies data from S3 buckets into redshift staging tables
• ./airflow/plugins/operators/load_dimension.py: Operator that loads data from redshift staging tables into dimensional tables
• ./airflow/plugins/operators/load_fact.py: Operator that loads data from redshift staging tables into fact table
• ./airflow/plugins/operators/data_quality.py: Operator that validates data quality in redshift tables

Airflow Dag

Run Project Locally

• Install Airflow, create variable AIRFLOW_HOME and AIRFLOW_CONFIG with the appropiate paths, and place dags and plugins on airflor_home directory.
  • export AIRFLOW_HOME=~/airflow
  • pip install apache-airflow
  • pip install postgres
  • pip install gunicorn
  • pip install boto3
  • airflow version to check package version
• Initialize Airflow data base with airflow initdb, and open webserver with airflow webserver
  • copy code to Airflow DAG folder (airflow/dags)
  • airflow initdb to initialize database
  • airflow scheduler -D to run scheduler
  • airflow webserver -p 8080 -D inside of airflow folder
• Access the server http://localhost:8080 and create:
  • AWS Connection
    • Conn Id: aws_credentials
    • Conn Type: select Amazon Web Services
    • Login: your Access key ID from the IAM User credentials
    • Password: your Secret access key from the IAM User credentials
  • Redshift Connection
    • Conn Id: redshift
    • Conn Type: select Postgres
    • Host: the endpoint of your Redshift cluster, excluding the port at the end
    • Schema: Redshift database you want to connect to
    • Login: awsuser
    • Password: the password created when launching the Redshift cluster
    • Port: 5439