Update chapter 3 for final review.

.gitignore

@@ -143,6 +143,7 @@ dmypy.json
 
 # Data folders
 data
+logs
 
 # Pyenv
 .python-version

README.md

@@ -13,7 +13,7 @@ Overall, this repository is structured as follows:
 ├── README.md             # This readme.
 ├── chapters              # Code examples for each of the Chapters.
 ├── docker                # Supporting Docker image (containing Airflow).
-└── environment.yml
+└── requirements.txt
 ```
 
 The most interesting parts are probably the *docker* directory and the *chapter* directories under *chapters*.

chapters/chapter01/docker-compose.yml

@@ -33,6 +33,8 @@ services:
       - postgres
     ports:
       - "8080:8080"
+    volumes:
+      - ./logs:/opt/airflow/logs
     environment: *airflow_environment
     command: webserver
   scheduler:
@@ -42,5 +44,6 @@ services:
       - postgres
     volumes:
       - ./dags:/opt/airflow/dags
+      - ./logs:/opt/airflow/logs
     environment: *airflow_environment
     command: scheduler

chapters/chapter3/dags/chapter3_01_unscheduled.py → chapters/chapter03/dags/01_unscheduled.py

@@ -7,16 +7,14 @@
 from airflow.operators.python_operator import PythonOperator
 
 dag = DAG(
-    dag_id="chapter3_01_unscheduled",
-    start_date=datetime(2015, 6, 1),
-    schedule_interval=None,
+    dag_id="01_unscheduled", start_date=datetime(2015, 6, 1), schedule_interval=None
 )
 
 fetch_events = BashOperator(
     task_id="fetch_events",
     bash_command=(
         "mkdir -p /data/events && "
-        "curl -o /data/events.json https://events_api:5000/events"
+        "curl -o /data/events.json http://events_api:5000/events"
     ),
     dag=dag,
 )

chapters/chapter3/dags/chapter3_02_daily_schedule.py → chapters/chapter03/dags/02_daily_schedule.py

@@ -7,16 +7,17 @@
 from airflow.operators.python_operator import PythonOperator
 
 dag = DAG(
-    dag_id="chapter3_02_daily_schedule",
+    dag_id="02_daily_schedule",
     start_date=datetime(2019, 1, 1),
+    end_date=datetime(2019, 1, 5),
     schedule_interval="@daily",
 )
 
 fetch_events = BashOperator(
     task_id="fetch_events",
     bash_command=(
         "mkdir -p /data/events && "
-        "curl -o /data/events.json https://events_api:5000/events"
+        "curl -o /data/events.json http://events_api:5000/events"
     ),
     dag=dag,
 )
@@ -35,7 +36,7 @@ def _calculate_stats(input_path, output_path):
 calculate_stats = PythonOperator(
     task_id="calculate_stats",
     python_callable=_calculate_stats,
-    op_kwargs={"input_path": "data/events.json", "output_path": "data/stats.csv"},
+    op_kwargs={"input_path": "/data/events.json", "output_path": "/data/stats.csv"},
     dag=dag,
 )
 

chapters/chapter3/dags/chapter3_03_different_start_date.py → chapters/chapter03/dags/03_different_start_date.py

@@ -7,7 +7,7 @@
 from airflow.operators.python_operator import PythonOperator
 
 dag = DAG(
-    dag_id="chapter3_03_different_start_date",
+    dag_id="03_different_start_date",
     schedule_interval="@daily",
     start_date=dt.datetime(year=2019, month=1, day=1),
 )
@@ -16,7 +16,7 @@
     task_id="fetch_events",
     bash_command=(
         "mkdir -p /data/events && "
-        "curl -o /data/events.json https://events_api:5000/events"
+        "curl -o /data/events.json http://events_api:5000/events"
     ),
     dag=dag,
 )
@@ -35,7 +35,7 @@ def _calculate_stats(input_path, output_path):
 calculate_stats = PythonOperator(
     task_id="calculate_stats",
     python_callable=_calculate_stats,
-    op_kwargs={"input_path": "data/events.json", "output_path": "data/stats.csv"},
+    op_kwargs={"input_path": "/data/events.json", "output_path": "/data/stats.csv"},
     dag=dag,
 )
 

chapters/chapter3/dags/chapter3_04_with_end_date.py → chapters/chapter03/dags/04_with_end_date.py

@@ -7,7 +7,7 @@
 from airflow.operators.python_operator import PythonOperator
 
 dag = DAG(
-    dag_id="chapter3_04_with_end_date",
+    dag_id="04_with_end_date",
     schedule_interval="@daily",
     start_date=dt.datetime(year=2019, month=1, day=1),
     end_date=dt.datetime(year=2019, month=1, day=5),
@@ -17,7 +17,7 @@
     task_id="fetch_events",
     bash_command=(
         "mkdir -p /data/events && "
-        "curl -o /data/events.json https://events_api:5000/events"
+        "curl -o /data/events.json http://events_api:5000/events"
     ),
     dag=dag,
 )
@@ -36,7 +36,7 @@ def _calculate_stats(input_path, output_path):
 calculate_stats = PythonOperator(
     task_id="calculate_stats",
     python_callable=_calculate_stats,
-    op_kwargs={"input_path": "data/events.json", "output_path": "data/stats.csv"},
+    op_kwargs={"input_path": "/data/events.json", "output_path": "/data/stats.csv"},
     dag=dag,
 )
 

chapters/chapter3/dags/chapter3_05_time_delta_schedule.py → chapters/chapter03/dags/05_time_delta_schedule.py

@@ -8,7 +8,7 @@
 from airflow.operators.python_operator import PythonOperator
 
 dag = DAG(
-    dag_id="chapter3_05_time_delta_schedule",
+    dag_id="05_time_delta_schedule",
     schedule_interval=timedelta(days=3),
     start_date=dt.datetime(year=2019, month=1, day=1),
     end_date=dt.datetime(year=2019, month=1, day=5),
@@ -18,7 +18,7 @@
     task_id="fetch_events",
     bash_command=(
         "mkdir -p /data/events && "
-        "curl -o /data/events.json https://events_api:5000/events"
+        "curl -o /data/events.json http://events_api:5000/events"
     ),
     dag=dag,
 )
@@ -37,7 +37,7 @@ def _calculate_stats(input_path, output_path):
 calculate_stats = PythonOperator(
     task_id="calculate_stats",
     python_callable=_calculate_stats,
-    op_kwargs={"input_path": "data/events.json", "output_path": "data/stats.csv"},
+    op_kwargs={"input_path": "/data/events.json", "output_path": "/data/stats.csv"},
     dag=dag,
 )
 

chapters/chapter3/dags/chapter3_06_query_with_dates.py → chapters/chapter03/dags/06_query_with_dates.py

@@ -8,7 +8,7 @@
 from airflow.operators.python_operator import PythonOperator
 
 dag = DAG(
-    dag_id="chapter3_06_query_with_dates",
+    dag_id="06_query_with_dates",
     schedule_interval=timedelta(days=3),
     start_date=dt.datetime(year=2019, month=1, day=1),
     end_date=dt.datetime(year=2019, month=1, day=5),
@@ -40,7 +40,7 @@ def _calculate_stats(input_path, output_path):
 calculate_stats = PythonOperator(
     task_id="calculate_stats",
     python_callable=_calculate_stats,
-    op_kwargs={"input_path": "data/events.json", "output_path": "data/stats.csv"},
+    op_kwargs={"input_path": "/data/events.json", "output_path": "/data/stats.csv"},
     dag=dag,
 )
 

chapters/chapter3/dags/chapter3_07_templated_query.py → chapters/chapter03/dags/07_templated_query.py

@@ -8,7 +8,7 @@
 from airflow.operators.python_operator import PythonOperator
 
 dag = DAG(
-    dag_id="chapter3_07_templated_query",
+    dag_id="07_templated_query",
     schedule_interval=timedelta(days=3),
     start_date=dt.datetime(year=2019, month=1, day=1),
     end_date=dt.datetime(year=2019, month=1, day=5),
@@ -40,7 +40,7 @@ def _calculate_stats(input_path, output_path):
 calculate_stats = PythonOperator(
     task_id="calculate_stats",
     python_callable=_calculate_stats,
-    op_kwargs={"input_path": "data/events.json", "output_path": "data/stats.csv"},
+    op_kwargs={"input_path": "/data/events.json", "output_path": "/data/stats.csv"},
     dag=dag,
 )
 

chapters/chapter3/dags/chapter3_08_templated_query_ds.py → chapters/chapter03/dags/08_templated_query_ds.py

@@ -8,7 +8,7 @@
 from airflow.operators.python_operator import PythonOperator
 
 dag = DAG(
-    dag_id="chapter3_08_templated_query_ds",
+    dag_id="08_templated_query_ds",
     schedule_interval=timedelta(days=3),
     start_date=dt.datetime(year=2019, month=1, day=1),
     end_date=dt.datetime(year=2019, month=1, day=5),
@@ -40,7 +40,7 @@ def _calculate_stats(input_path, output_path):
 calculate_stats = PythonOperator(
     task_id="calculate_stats",
     python_callable=_calculate_stats,
-    op_kwargs={"input_path": "data/events.json", "output_path": "data/stats.csv"},
+    op_kwargs={"input_path": "/data/events.json", "output_path": "/data/stats.csv"},
     dag=dag,
 )
 

chapters/chapter3/dags/chapter3_09_templated_path.py → chapters/chapter03/dags/09_templated_path.py

@@ -8,7 +8,7 @@
 from airflow.operators.python_operator import PythonOperator
 
 dag = DAG(
-    dag_id="chapter3_09_templated_path",
+    dag_id="09_templated_path",
     schedule_interval=timedelta(days=3),
     start_date=dt.datetime(year=2019, month=1, day=1),
     end_date=dt.datetime(year=2019, month=1, day=5),
@@ -43,8 +43,8 @@ def _calculate_stats(**context):
     task_id="calculate_stats",
     python_callable=_calculate_stats,
     templates_dict={
-        "input_path": "data/events/{{ds}}.json",
-        "output_path": "data/stats/{{ds}}.csv",
+        "input_path": "/data/events/{{ds}}.json",
+        "output_path": "/data/stats/{{ds}}.csv",
     },
     provide_context=True,
     dag=dag,

chapters/chapter3/dags/chapter3_10_full_example.py → chapters/chapter03/dags/10_full_example.py

@@ -8,7 +8,7 @@
 from airflow.operators.python_operator import PythonOperator
 
 dag = DAG(
-    dag_id="chapter3_10_full_example",
+    dag_id="10_full_example",
     schedule_interval=timedelta(days=3),
     start_date=dt.datetime(year=2019, month=1, day=1),
     end_date=dt.datetime(year=2019, month=1, day=5),

chapters/chapter03/docker-compose.yml

@@ -0,0 +1,55 @@
+version: '3.7'
+# ====================================== AIRFLOW ENVIRONMENT VARIABLES =======================================
+x-environment: &airflow_environment
+  - AIRFLOW__CORE__EXECUTOR=LocalExecutor
+  - AIRFLOW__CORE__LOAD_DEFAULT_CONNECTIONS=False
+  - AIRFLOW__CORE__LOAD_EXAMPLES=False
+  - AIRFLOW__CORE__SQL_ALCHEMY_CONN=postgresql://airflow:airflow@postgres:5432/airflow
+  - AIRFLOW__CORE__STORE_DAG_CODE=True
+  - AIRFLOW__CORE__STORE_SERIALIZED_DAGS=True
+  - AIRFLOW__WEBSERVER__EXPOSE_CONFIG=True
+  - AIRFLOW__WEBSERVER__RBAC=False
+x-airflow-image: &airflow_image apache/airflow:1.10.12-python3.8
+# ====================================== /AIRFLOW ENVIRONMENT VARIABLES ======================================
+services:
+  postgres:
+    image: postgres:12-alpine
+    environment:
+      - POSTGRES_USER=airflow
+      - POSTGRES_PASSWORD=airflow
+      - POSTGRES_DB=airflow
+    ports:
+      - "5432:5432"
+  init:
+    image: *airflow_image
+    depends_on:
+      - postgres
+    environment: *airflow_environment
+    command: upgradedb
+  webserver:
+    image: *airflow_image
+    restart: always
+    depends_on:
+      - postgres
+    ports:
+      - "8080:8080"
+    volumes:
+      - ./logs:/opt/airflow/logs
+    environment: *airflow_environment
+    command: webserver
+  scheduler:
+    image: *airflow_image
+    restart: always
+    depends_on:
+      - postgres
+    volumes:
+      - ./dags:/opt/airflow/dags
+      - ./logs:/opt/airflow/logs
+      - ./data:/data
+    environment: *airflow_environment
+    command: scheduler
+  events_api:
+    build: ./api
+    image: manning-airflow-events-api:latest
+    ports:
+      - "5000:5000"

chapters/chapter03/readme.md

@@ -0,0 +1,30 @@
+# Chapter 3
+
+Code accompanying Chapter 3 of the book 'Data pipelines with Apache Airflow'.
+
+## Contents
+
+This code example contains the following DAGs:
+
+- 01_unscheduled.py - Initial DAG without schedule.
+- 02_daily_schedule.py - Same DAG following a daily schedule.
+- 03_different_start_date.py - DAG with adjusted start date.
+- 04_with_end_date.py - Modified DAG with an end date.
+- 05_time_delta_schedule.py - DAG that uses a timedelta for the schedule interval.
+- 06_query_with_dates.py - DAG including hard-coded dates in the query.
+- 07_templated_query.py - Replaces hard-coded dates with templated execution dates.
+- 08_templated_query_ds.py - Uses shorthands for the templated execution dates.
+- 09_templated_path.py - Uses templating for the file paths as well.
+- 10_full_example.py - Filly completed example, including 'sending' of statistics.
+
+## Usage
+
+To get started with the code examples, start Airflow in docker using the following command:
+
+    docker-compose up -d --build
+
+Wait for a few seconds and you should be able to access the examples at http://localhost:8080/.
+
+To stop running the examples, run the following command:
+
+    docker-compose down