Update chapter 7 for final review (becomes chapter 8). (#20)

* Updated chapter 7, renamed to chapter 8.

* Rename src to avoid confusion.

* Update compose to use official Airflow image.

* Fix linting issues.

Co-authored-by: Julian de Ruiter <[email protected]>

Author: Julian de Ruiter

chapters/chapter03/docker-compose.yml

@@ -50,6 +50,6 @@ services:
     command: scheduler
   events_api:
     build: ./api
-    image: manning-airflow-events-api:latest
+    image: manning-airflow/events-api:latest
     ports:
       - "5000:5000"

chapters/chapter08/api/Dockerfile

@@ -0,0 +1,12 @@
+FROM python:3.8-slim
+
+COPY requirements.txt /tmp/requirements.txt
+RUN pip install -r /tmp/requirements.txt && rm -f /tmp/requirements.txt
+
+COPY app.py fetch_ratings.py /
+RUN python /fetch_ratings.py --output_path /ratings.csv
+
+EXPOSE 5000
+
+ENTRYPOINT ["python"]
+CMD ["/app.py"]

chapters/chapter7/api/app.py renamed to chapters/chapter08/api/app.py

@@ -1,8 +1,6 @@
-import datetime as dt
 import os
 import time
 
-import numpy as np
 import pandas as pd
 
 from flask import Flask, jsonify, request
@@ -12,43 +10,20 @@
 DEFAULT_ITEMS_PER_PAGE = 100
 
 
-def _read_ratings(file_path, shift_ts=True):
+def _read_ratings(file_path):
     ratings = pd.read_csv(file_path)
 
     # Subsample dataset.
-    ratings = ratings.sample(n=100000)
+    ratings = ratings.sample(n=100000, random_state=0)
 
     # Sort by ts, user, movie for convenience.
     ratings = ratings.sort_values(by=["timestamp", "userId", "movieId"])
 
-    # Replace timestamps with timestamps from
-    # within the last month.
-    if shift_ts:
-        today = dt.datetime.now().date()
-
-        ratings["timestamp"] = _random_timestamps(
-            start_date=today + dt.timedelta(days=-30),
-            end_date=today,
-            size=ratings.shape[0],
-        )
-
     return ratings
 
 
-def _random_timestamps(start_date=None, end_date=None, size=100):
-    """Generates random timestamps (in seconds) between given start/end dates."""
-
-    def _date_to_datetime(date):
-        return dt.datetime.combine(date, dt.datetime.min.time())
-
-    start_ts = int(_date_to_datetime(start_date).timestamp())
-    end_ts = int(_date_to_datetime(end_date).timestamp())
-
-    return np.random.randint(low=start_ts, high=end_ts, size=size)
-
-
 app = Flask(__name__)
-app.config["ratings"] = _read_ratings("/ratings.csv", shift_ts=True)
+app.config["ratings"] = _read_ratings("/ratings.csv")
 
 auth = HTTPBasicAuth()
 users = {os.environ["API_USER"]: generate_password_hash(os.environ["API_PASSWORD"])}

chapters/chapter08/api/fetch_ratings.py

@@ -0,0 +1,55 @@
+import logging
+from pathlib import Path
+import tempfile
+from urllib.request import urlretrieve
+import zipfile
+
+import click
+import pandas as pd
+
+logging.basicConfig(
+    format="[%(asctime)-15s] %(levelname)s - %(message)s", level=logging.INFO
+)
+
+
+@click.command()
+@click.option("--start_date", default="2019-01-01", type=click.DateTime())
+@click.option("--end_date", default="2020-01-01", type=click.DateTime())
+@click.option("--output_path", required=True)
+def main(start_date, end_date, output_path):
+    """Script for fetching movielens ratings within a given date range."""
+
+    logging.info("Fetching ratings...")
+    ratings = fetch_ratings()
+
+    # Subset to expected range.
+    logging.info(f"Filtering for dates {start_date} - {end_date}...")
+    ts_parsed = pd.to_datetime(ratings["timestamp"], unit="s")
+    ratings = ratings.loc[(ts_parsed >= start_date) & (ts_parsed < end_date)]
+
+    logging.info(f"Writing ratings to '{output_path}'...")
+    ratings.to_csv(output_path, index=False)
+
+
+def fetch_ratings():
+    """Fetches ratings from the given URL."""
+
+    url = "http://files.grouplens.org/datasets/movielens/ml-25m.zip"
+
+    with tempfile.TemporaryDirectory() as tmp_dir:
+        tmp_path = Path(tmp_dir, "download.zip")
+        logging.info(f"Downloading zip file from {url}")
+        urlretrieve(url, tmp_path)
+
+        with zipfile.ZipFile(tmp_path) as zip_:
+            logging.info(f"Downloaded zip file with contents: {zip_.namelist()}")
+
+            logging.info("Reading ml-25m/ratings.csv from zip file")
+            with zip_.open("ml-25m/ratings.csv") as file_:
+                ratings = pd.read_csv(file_)
+
+    return ratings
+
+
+if __name__ == "__main__":
+    main()

chapters/chapter7/api/requirements.txt renamed to chapters/chapter08/api/requirements.txt

@@ -1,3 +1,4 @@
 Flask==1.1.1
 pandas==0.25.2
 Flask-HTTPAuth==3.3.0
+click

chapters/chapter7/dags/movielens_python_operator.py renamed to chapters/chapter08/dags/01_movielens_python_operator.py

@@ -1,11 +1,12 @@
+import datetime as dt
 import logging
 import json
 import os
 
 import pandas as pd
 import requests
 
-from airflow import DAG, utils as airflow_utils
+from airflow import DAG
 from airflow.operators.python_operator import PythonOperator
 
 from custom.ranking import rank_movies_by_rating
@@ -71,9 +72,10 @@ def _get_with_pagination(session, url, params, batch_size=100):
 
 
 with DAG(
-    dag_id="chapter7_movielens_python_operator",
+    dag_id="01_movielens_python_operator",
     description="Fetches ratings from the Movielens API using the Python Operator.",
-    start_date=airflow_utils.dates.days_ago(7),
+    start_date=dt.datetime(2019, 1, 1),
+    end_date=dt.datetime(2019, 1, 10),
     schedule_interval="@daily",
 ) as dag:
 
@@ -117,8 +119,12 @@ def _rank_movies(templates_dict, min_ratings=2, **_):
         output_path = templates_dict["output_path"]
 
         ratings = pd.read_json(input_path)
-
         ranking = rank_movies_by_rating(ratings, min_ratings=min_ratings)
+
+        # Make sure output directory exists.
+        output_dir = os.path.dirname(output_path)
+        os.makedirs(output_dir, exist_ok=True)
+
         ranking.to_csv(output_path, index=True)
 
     rank_movies = PythonOperator(

chapters/chapter7/dags/movielens_custom_hook.py renamed to chapters/chapter08/dags/02_movielens_custom_hook.py

@@ -1,17 +1,19 @@
+import datetime as dt
 import logging
 import json
 import os
 
-from airflow import DAG, utils as airflow_utils
+from airflow import DAG
 from airflow.operators.python_operator import PythonOperator
 
 from custom.hooks import MovielensHook
 
 
 with DAG(
-    dag_id="chapter7_movielens_custom_hook",
+    dag_id="02_movielens_custom_hook",
     description="Fetches ratings from the Movielens API using a custom hook.",
-    start_date=airflow_utils.dates.days_ago(7),
+    start_date=dt.datetime(2019, 1, 1),
+    end_date=dt.datetime(2019, 1, 10),
     schedule_interval="@daily",
 ) as dag:
 

chapters/chapter7/dags/movielens_custom_operator.py renamed to chapters/chapter08/dags/03_movielens_custom_operator.py

@@ -1,12 +1,15 @@
-from airflow import DAG, utils as airflow_utils
+import datetime as dt
+
+from airflow import DAG
 
 from custom.operators import MovielensFetchRatingsOperator
 
 
 with DAG(
-    dag_id="chapter7_movielens_custom_operator",
+    dag_id="03_movielens_custom_operator",
     description="Fetches ratings from the Movielens API using a custom operator.",
-    start_date=airflow_utils.dates.days_ago(7),
+    start_date=dt.datetime(2019, 1, 1),
+    end_date=dt.datetime(2019, 1, 10),
     schedule_interval="@daily",
 ) as dag:
     MovielensFetchRatingsOperator(

chapters/chapter7/dags/movielens_custom_sensor.py renamed to chapters/chapter08/dags/04_movielens_custom_sensor.py

@@ -1,12 +1,15 @@
-from airflow import DAG, utils as airflow_utils
+import datetime as dt
+
+from airflow import DAG
 
 from custom.operators import MovielensFetchRatingsOperator
 from custom.sensors import MovielensRatingsSensor
 
 with DAG(
-    dag_id="chapter7_movielens_sensor",
+    dag_id="04_movielens_sensor",
     description="Fetches ratings from the Movielens API, with a custom sensor.",
-    start_date=airflow_utils.dates.days_ago(7),
+    start_date=dt.datetime(2019, 1, 1),
+    end_date=dt.datetime(2019, 1, 10),
     schedule_interval="@daily",
 ) as dag:
     wait_for_ratings = MovielensRatingsSensor(