Merge pull request #23 from godatadriven/kg/update_ch12

Switch chapters 12 and 14 around

Author: krisgeus

chapter12/.env

@@ -0,0 +1,2 @@
+AIRFLOW_UID=1000
+AIRFLOW_GID=0

chapter14/Makefile renamed to chapter12/Makefile

@@ -1,38 +1,79 @@
-# Chapter 12
+# Chapter 14
 
-Code accompanying Chapter 12 of the book [Data Pipelines with Apache Airflow](https://www.manning.com/books/data-pipelines-with-apache-airflow).
+Code accompanying Chapter 14 of the book [Data Pipelines with Apache Airflow](https://www.manning.com/books/data-pipelines-with-apache-airflow).
 
-## Contents
+Much of this use case was based on the ideas in https://toddwschneider.com/posts/taxi-vs-citi-bike-nyc, where
+the fastest method of transportation (Citi Bikes or Yellow Taxis) between two NYC neighborhoods at given times
+and days is determines. We will productionize this use case into an Airflow workflow. Two (mocked) real-time
+services are created for this purpose:
 
-This folder contains DAGs from Chapter 12. Topics covered are monitoring, logging, scaling horizontal, etc. An
-accompanying Docker Compose setup was built for demonstration purposes. This includes:
+1. A service providing Citi Bike rides
+1. A service providing NYC Yellow Taxi rides
 
-- Airflow (webserver, scheduler, and Celery workers)
-- PostgreSQL database for Airflow metastore
-- Redis for Celery queue
-- Flower, a Celery monitoring tool
-- Prometheus, for scraping and storing metrics
-- Grafana, for visualizing metrics
-- And a Redis & StatsD exporter to expose metrics
+An Airflow DAG will periodically scrape these services, wrangle the data, and send the results into a Postgres
+database serving a tiny website, displaying which transportation method is fastest between two NYC
+neighborhoods.
 
-Given the number of services, this can become a bit resource-heavy on your machine.
+The real data is only available in batches of months/years. Therefore we provide two APIs to mimic "live"
+systems. The system can be brought online with the `docker-compose.yml` provided with this repository:
 
-Unfortunately, not everything can be scripted/pre-initialized, especially in Grafana. Therefore, you must add
-Prometheus as a datasource and create a dashboard yourself.
+```bash
+docker-compose up -d
+```
 
-## Usage
+If all processing goes well, the final result is visible on http://localhost:8083.
 
-To get started with the code examples, start Airflow with Docker Compose with the following command:
+Ports on which services are available:
 
-```bash
-docker-compose up -d
+- http://localhost:5432: Airflow Postgres DB (`airflow`/`airflow`)
+- http://localhost:5433: NYC Taxi Postgres DB (`taxi`/`ridetlc`)
+- http://localhost:5434: Citi Bike Postgres DB (`citi`/`cycling`)
+- http://localhost:5435: NYC Transportation results Postgres DB (`nyc`/`tr4N5p0RT4TI0N`)
+- http://localhost:8080: Airflow webserver (`airflow`/`airflow`)
+- http://localhost:8081: NYC Taxi static file server
+- http://localhost:8082: Citi Bike API (`citibike`/`cycling`)
+- http://localhost:8083: NYC Transportation API
+- http://localhost:9000: MinIO (`AKIAIOSFODNN7EXAMPLE`/`wJalrXUtnFEMI/K7MDENG/bPxRfiCYEXAMPLEKEY`)
+
+## More information
+
+### Taxi dataset
+
+The taxi dataset refers to the "NYC Taxi and Limousine Commission (TLC) Trip Record Data", available here:
+https://www1.nyc.gov/site/tlc/about/tlc-trip-record-data.page.
+
+A snippet of the data:
+
+```csv
+VendorID,tpep_pickup_datetime,tpep_dropoff_datetime,passenger_count,trip_distance,RatecodeID,store_and_fwd_flag,PULocationID,DOLocationID,payment_type,fare_amount,extra,mta_tax,tip_amount,tolls_amount,improvement_surcharge,total_amount,congestion_surcharge
+1,2019-01-01 00:46:40,2019-01-01 00:53:20,1,1.50,1,N,151,239,1,7,0.5,0.5,1.65,0,0.3,9.95,
+1,2019-01-01 00:59:47,2019-01-01 01:18:59,1,2.60,1,N,239,246,1,14,0.5,0.5,1,0,0.3,16.3,
+2,2018-12-21 13:48:30,2018-12-21 13:52:40,3,.00,1,N,236,236,1,4.5,0.5,0.5,0,0,0.3,5.8,
+2,2018-11-28 15:52:25,2018-11-28 15:55:45,5,.00,1,N,193,193,2,3.5,0.5,0.5,0,0,0.3,7.55,
+2,2018-11-28 15:56:57,2018-11-28 15:58:33,5,.00,2,N,193,193,2,52,0,0.5,0,0,0.3,55.55,
+2,2018-11-28 16:25:49,2018-11-28 16:28:26,5,.00,1,N,193,193,2,3.5,0.5,0.5,0,5.76,0.3,13.31,
+2,2018-11-28 16:29:37,2018-11-28 16:33:43,5,.00,2,N,193,193,2,52,0,0.5,0,0,0.3,55.55,
+1,2019-01-01 00:21:28,2019-01-01 00:28:37,1,1.30,1,N,163,229,1,6.5,0.5,0.5,1.25,0,0.3,9.05,
+1,2019-01-01 00:32:01,2019-01-01 00:45:39,1,3.70,1,N,229,7,1,13.5,0.5,0.5,3.7,0,0.3,18.5,
 ```
 
-The webserver initializes a few things, so wait for a few seconds, and you should be able to access the
-Airflow webserver at http://localhost:8080.
+Data dictionary: https://www1.nyc.gov/assets/tlc/downloads/pdf/data_dictionary_trip_records_yellow.pdf
 
-To stop running the examples, run the following command:
+### Citi Bike dataset
 
-```bash
-docker-compose down -v
+https://www.citibikenyc.com/system-data
+
+A snippet of the data:
+
+```csv
+"tripduration","starttime","stoptime","start station id","start station name","start station latitude","start station longitude","end station id","end station name","end station latitude","end station longitude","bikeid","usertype","birth year","gender"
+201,"2019-01-01 03:09:09.7110","2019-01-01 03:12:30.8790",3183,"Exchange Place",40.7162469,-74.0334588,3214,"Essex Light Rail",40.7127742,-74.0364857,29612,"Subscriber",1993,1
+505,"2019-01-01 05:18:00.1060","2019-01-01 05:26:25.9050",3183,"Exchange Place",40.7162469,-74.0334588,3638,"Washington St",40.7242941,-74.0354826,29213,"Subscriber",1972,2
+756,"2019-01-01 10:36:33.3400","2019-01-01 10:49:10.2600",3183,"Exchange Place",40.7162469,-74.0334588,3192,"Liberty Light Rail",40.7112423,-74.0557013,26164,"Subscriber",1985,1
+1575,"2019-01-01 12:43:38.6430","2019-01-01 13:09:54.5280",3183,"Exchange Place",40.7162469,-74.0334588,3638,"Washington St",40.7242941,-74.0354826,29672,"Customer",1969,0
+1566,"2019-01-01 12:43:39.6010","2019-01-01 13:09:46.5100",3183,"Exchange Place",40.7162469,-74.0334588,3638,"Washington St",40.7242941,-74.0354826,29522,"Customer",1969,0
+737,"2019-01-01 12:56:53.2040","2019-01-01 13:09:11.0400",3183,"Exchange Place",40.7162469,-74.0334588,3205,"JC Medical Center",40.71653978099194,-74.0496379137039,29447,"Subscriber",1993,1
+917,"2019-01-01 13:03:44.7760","2019-01-01 13:19:02.7690",3183,"Exchange Place",40.7162469,-74.0334588,3277,"Communipaw & Berry Lane",40.71435836870427,-74.06661093235016,29299,"Subscriber",1986,1
+3248,"2019-01-01 13:12:03.1280","2019-01-01 14:06:12.0400",3183,"Exchange Place",40.7162469,-74.0334588,3196,"Riverview Park",40.7443187,-74.0439909,29495,"Subscriber",1992,1
+3168,"2019-01-01 13:13:12.0450","2019-01-01 14:06:00.4110",3183,"Exchange Place",40.7162469,-74.0334588,3196,"Riverview Park",40.7443187,-74.0439909,26312,"Customer",1969,0
 ```

chapter12/README.md

@@ -2,16 +2,19 @@ version: '3.7'
 
 # ====================================== AIRFLOW ENVIRONMENT VARIABLES =======================================
 x-environment: &airflow_environment
-  - AIRFLOW__CELERY__BROKER_URL=redis://redis:6379/1
-  - AIRFLOW__CELERY__RESULT_BACKEND=db+postgresql://airflow:airflow@postgres:5432/airflow
-  - AIRFLOW__CORE__EXECUTOR=CeleryExecutor
+  - AIRFLOW__CORE__DAGS_ARE_PAUSED_AT_CREATION=False
+  - AIRFLOW__CORE__EXECUTOR=LocalExecutor
+  - AIRFLOW__CORE__FERNET_KEY=hCRoPUYBO27QiEg1MRu5hSjLG7yNd8y8XKlm-8kRlkQ=
   - AIRFLOW__CORE__LOAD_DEFAULT_CONNECTIONS=False
   - AIRFLOW__CORE__LOAD_EXAMPLES=False
   - AIRFLOW__CORE__SQL_ALCHEMY_CONN=postgresql://airflow:airflow@postgres:5432/airflow
-  - AIRFLOW__METRICS__STATSD_HOST=statsd_exporter
-  - AIRFLOW__METRICS__STATSD_ON=True
-  - AIRFLOW__METRICS__STATSD_PORT=9125
+  - AIRFLOW__CORE__STORE_DAG_CODE=True
+  - AIRFLOW__CORE__STORE_SERIALIZED_DAGS=True
   - AIRFLOW__WEBSERVER__EXPOSE_CONFIG=True
+  - AIRFLOW_CONN_S3=s3://@?host=http://minio:9000&aws_access_key_id=AKIAIOSFODNN7EXAMPLE&aws_secret_access_key=wJalrXUtnFEMI/K7MDENG/bPxRfiCYEXAMPLEKEY
+  - AIRFLOW_CONN_CITIBIKE=http://citibike:cycling@citibike_api:5000
+  - AIRFLOW_CONN_TAXI=http://taxi_fileserver
+  - AIRFLOW_CONN_RESULT_DB=postgresql://nyc:tr4N5p0RT4TI0N@result_db:5432/nyctransportation
 
 x-airflow-image: &airflow_image apache/airflow:2.0.0-python3.8
 # ====================================== /AIRFLOW ENVIRONMENT VARIABLES ======================================
@@ -26,99 +29,139 @@ services:
     ports:
       - "5432:5432"
 
-  init:
-    image: *airflow_image
+  initdb_adduser:
+    build:
+      context: ./services/airflow
+      args:
+        AIRFLOW_BASE_IMAGE: *airflow_image
     depends_on:
       - postgres
     environment: *airflow_environment
     entrypoint: /bin/bash
     command: -c 'airflow db init && airflow users create --username admin --password admin --firstname Anonymous --lastname Admin --role Admin --email [email protected]'
 
   webserver:
-    image: *airflow_image
+    build:
+      context: ./services/airflow
+      args:
+        AIRFLOW_BASE_IMAGE: *airflow_image
     restart: always
     depends_on:
       - postgres
-    ports:
-      - "8080:8080"
     volumes:
       - logs:/opt/airflow/logs
+    ports:
+      - "8080:8080"
     environment: *airflow_environment
     command: webserver
 
   scheduler:
-    image: *airflow_image
+    build:
+      context: ./services/airflow
+      args:
+        AIRFLOW_BASE_IMAGE: *airflow_image
     restart: always
     depends_on:
       - postgres
     volumes:
-      - logs:/opt/airflow/logs
       - ./dags:/opt/airflow/dags
+      - ./src:/opt/airflow/nyctransport/src
+      - ./setup.py:/opt/airflow/nyctransport/setup.py
+      - logs:/opt/airflow/logs
     environment: *airflow_environment
-    command: scheduler
+    entrypoint: ["/bin/sh"]
+    command: ["-c", "pip install --user -e /opt/airflow/nyctransport && airflow scheduler"]
 
-  # docker-compose -f docker-compose-celeryexecutor.yml up --scale worker=3 -d
-  worker:
-    image: *airflow_image
-    restart: always
+  taxi_fileserver:
+    build:
+      context: ./services/taxi_fileserver
+    environment:
+      - POSTGRES_HOST=taxi_db
+      - POSTGRES_PORT=5433
+      - POSTGRES_USERNAME=taxi
+      - POSTGRES_PASSWORD=ridetlc
+      - POSTGRES_DATABASE=tlctriprecords
+      - DATA_YEAR=2019
     depends_on:
-      - scheduler
-    volumes:
-      - logs:/opt/airflow/logs
-      - ./dags:/opt/airflow/dags
-    environment: *airflow_environment
-    command: celery worker
+      - taxi_db
+    ports:
+      - "8081:80"
 
-  flower:
-    image: *airflow_image
-    restart: always
+  taxi_db:
+    build:
+      context: ./services/taxi_db
+    ports:
+      - "5433:5432"
+
+  citibike_api:
+    build:
+      context: ./services/citibike_api
+    environment:
+      - POSTGRES_HOST=citibike_db
+      - POSTGRES_PORT=5434
+      - POSTGRES_USERNAME=citi
+      - POSTGRES_PASSWORD=cycling
+      - POSTGRES_DATABASE=citibike
+      - DATA_YEAR=2019
     depends_on:
-      - worker
+      - citibike_db
     ports:
-      - "5555:5555"
-    environment: *airflow_environment
-    command: celery flower
+      - "8082:5000"
 
-  statsd_exporter:
-    image: prom/statsd-exporter:v0.18.0
-    restart: always
-    volumes:
-      - ./files/statsd_mapping.yml:/tmp/statsd_mapping.yml
+  citibike_db:
+    build:
+      context: ./services/citibike_db
     ports:
-      - "9102:9102"
-      - "9125:9125/udp"
-    command: --statsd.mapping-config=/tmp/statsd_mapping.yml
+      - "5434:5432"
 
-  prometheus:
-    image: prom/prometheus:v2.22.0
-    restart: always
+  minio:
+    image: minio/minio:RELEASE.2020-06-22T03-12-50Z
     volumes:
-      - ./files/prometheus.yml:/etc/prometheus/prometheus.yml
+      - s3:/data
     ports:
-      - "9090:9090"
-    command:
-      - --web.enable-admin-api
-      - --web.enable-lifecycle
-      # Flags below are defaults, but must be added explicitly, otherwise would be overridden by flags above
-      - --config.file=/etc/prometheus/prometheus.yml
-      - --storage.tsdb.path=/prometheus
-      - --web.console.libraries=/usr/share/prometheus/console_libraries
-      - --web.console.templates=/usr/share/prometheus/consoles
+      - "9000:9000"
+    command: server /data
+    environment:
+      MINIO_ACCESS_KEY: AKIAIOSFODNN7EXAMPLE
+      MINIO_SECRET_KEY: wJalrXUtnFEMI/K7MDENG/bPxRfiCYEXAMPLEKEY
 
-  grafana:
-    image: grafana/grafana:7.2.1
-    restart: always
-    ports:
-      - "3000:3000"
+  minio_init:
+    image: minio/mc:RELEASE.2020-06-20T00-18-43Z
+    depends_on:
+      - minio
+    entrypoint: >
+      /bin/sh -c "
+      while ! nc -z minio 9000; do echo 'Waiting 1 sec for MinIO to be healthy...' && sleep 1; done;
+      echo 'MinIO is available.';
+      while ! /usr/bin/mc config host add minio http://minio:9000 AKIAIOSFODNN7EXAMPLE wJalrXUtnFEMI/K7MDENG/bPxRfiCYEXAMPLEKEY; do echo 'MinIO not up and running yet...' && sleep 1; done;
+      echo 'Added mc host config.';
+      /usr/bin/mc mb minio/datalake;
+      exit 0;
+      "
 
-  redis:
-    image: redis:5-alpine
+  result_db:
+    image: postgres:13-alpine
+    environment:
+      - POSTGRES_USER=nyc
+      - POSTGRES_PASSWORD=tr4N5p0RT4TI0N
+      - POSTGRES_DB=nyctransportation
+    volumes:
+    - ./services/result_db/create_tables.sql:/docker-entrypoint-initdb.d/create_tables.sql
+    ports:
+      - "5435:5432"
 
-  redis_exporter:
-    image: oliver006/redis_exporter:v1.5.2-alpine
+  nyc_transportation_api:
+    build:
+      context: ./services/nyc_transportation_api
+    environment:
+      - POSTGRES_HOST=result_db
+      - POSTGRES_PORT=5432
+      - POSTGRES_USERNAME=nyc
+      - POSTGRES_PASSWORD=tr4N5p0RT4TI0N
+      - POSTGRES_DATABASE=nyctransportation
     ports:
-      - "9121:9121"
-    command: --redis.addr=redis://redis:6379
+      - "8083:5000"
 
 volumes:
   logs:
+  s3: