Lesson 2 - The Ingest Task

Welcome to the Ingest Task

Lesson 1’s ingest_stub printed a sentence and returned a placeholder string. This lesson replaces it with the real thing: a task that reads a genuine slice of CityFlow’s actual taxi data and writes it somewhere the rest of the pipeline can pick it up from. The scale matters here — this isn’t the 200-500 row sample Modules 2 and 4 used, it’s a real day carved out of a ~50MB parquet file covering the whole month.

By the end of this lesson, you will be able to:

Read a real parquet file with pandas and filter it to a real date range
Explain why ingest writes its output to a file instead of just returning a DataFrame in memory
Trigger a real ingest task and confirm its output two independent ways

CityFlow’s Real January Data

yellow_tripdata_2024-01.parquet is the real NYC taxi trip record file for January 2024 — the same public dataset this whole course has drawn from, at the scale a real month actually is:

import pandas as pd

df = pd.read_parquet("yellow_tripdata_2024-01.parquet", columns=["tpep_pickup_datetime"])
print(f"{len(df):,} total real rows in the January 2024 file")

2,964,624 total real rows in the January 2024 file

Nearly 3 million real rows for one month. This lesson’s demo works with one real day out of that file — 2024-01-01 — and the guided project in Lesson 5 scales the same task up to three real days. Either way, it’s real data sliced by a real date filter, never a hand-picked sample file.

The Real Ingest Task

# gate: skip
"""CityFlow's real ingest task, on its own (Module 5 Lesson 2).

Reads a real slice of CityFlow's actual January 2024 taxi data straight off
the ~50MB parquet file (not a 200-500 row sample like Modules 2 and 4 used)
and writes it to a CSV the next stage can pick up. This lesson's demo scale
is one real day (2024-01-01); Lesson 5's guided project scales the exact
same task up to three real days."""
from datetime import datetime

from airflow.decorators import dag, task

PARQUET_PATH = "/opt/airflow/dags/data/yellow_tripdata_2024-01.parquet"
RAW_TRIPS_PATH = "/opt/airflow/dags/data/work/etl_ingest_demo/raw_trips.csv"
INGEST_START = "2024-01-01"
INGEST_END_EXCLUSIVE = "2024-01-02"  # one real day for this lesson's demo
KEEP_COLUMNS = [
    "tpep_pickup_datetime", "tpep_dropoff_datetime", "passenger_count",
    "trip_distance", "PULocationID", "DOLocationID", "payment_type",
    "fare_amount", "tip_amount", "total_amount",
]


@dag(
    dag_id="cityflow_etl_ingest_demo",
    schedule=None,
    start_date=datetime(2026, 1, 1),
    catchup=False,
    tags=["cityflow", "module-5", "ingest"],
)
def cityflow_etl_ingest_demo_dag():
    @task
    def ingest_taxi_trips() -> int:
        """Read the real parquet file, keep only real trips whose pickup
        falls in [INGEST_START, INGEST_END_EXCLUSIVE), and write them to a
        real CSV the next stage of the pipeline reads from -- no sample
        file, no synthetic rows."""
        import os

        import pandas as pd

        raw = pd.read_parquet(PARQUET_PATH, columns=["tpep_pickup_datetime"] + [
            c for c in KEEP_COLUMNS if c != "tpep_pickup_datetime"
        ])
        raw["tpep_pickup_datetime"] = pd.to_datetime(raw["tpep_pickup_datetime"])
        raw["tpep_dropoff_datetime"] = pd.to_datetime(raw["tpep_dropoff_datetime"])

        start = pd.Timestamp(INGEST_START)
        end = pd.Timestamp(INGEST_END_EXCLUSIVE)
        sliced = raw[(raw["tpep_pickup_datetime"] >= start) & (raw["tpep_pickup_datetime"] < end)]
        sliced = sliced[KEEP_COLUMNS]

        os.makedirs(os.path.dirname(RAW_TRIPS_PATH), exist_ok=True)
        sliced.to_csv(RAW_TRIPS_PATH, index=False)

        print(
            f"ingested {len(sliced)} real CityFlow trips "
            f"({INGEST_START} to {INGEST_END_EXCLUSIVE}, exclusive) "
            f"-> {RAW_TRIPS_PATH}"
        )
        return len(sliced)

    ingest_taxi_trips()


cityflow_etl_ingest_demo_dag()

Two things worth noticing that aren’t in Lesson 1’s stub: the date filter uses pd.Timestamp comparisons on a real parsed datetime64 column, not a string prefix match — and the task writes its result to a real file on disk (RAW_TRIPS_PATH) rather than returning the DataFrame itself. That second choice matters once tasks run as genuinely separate processes: XCom (the mechanism return values travel through between TaskFlow tasks) is meant for small values like the row count this task returns, not for handing an 81,013-row DataFrame from one task’s memory to the next task’s memory. A real file path is the real answer for real-sized data.

Triggered for Real: 81,013 Rows

# gate: skip
import json
import subprocess
import time

dag_id = "cityflow_etl_ingest_demo"

deadline = time.time() + 320
found = False
while time.time() < deadline:
    result = subprocess.run(
        ["docker", "exec", "airflow-local-airflow-scheduler-1", "airflow", "dags", "list", "--output", "json"],
        capture_output=True, text=True, check=True,
    )
    if any(d["dag_id"] == dag_id for d in json.loads(result.stdout)):
        found = True
        break
    time.sleep(5)
assert found, f"{dag_id} was never picked up by the scheduler"

subprocess.run(["docker", "exec", "airflow-local-airflow-scheduler-1", "airflow", "dags", "unpause", dag_id],
                capture_output=True, text=True, check=True)
run_id = f"lesson2_m5_verify_{int(time.time())}"
trigger = subprocess.run(
    ["docker", "exec", "airflow-local-airflow-scheduler-1", "airflow", "dags", "trigger", dag_id, "--run-id", run_id],
    capture_output=True, text=True,
)
if trigger.returncode != 0:
    subprocess.run(["docker", "exec", "airflow-local-airflow-scheduler-1", "airflow", "dags", "reserialize"],
                    capture_output=True, text=True, check=True)
    subprocess.run(
        ["docker", "exec", "airflow-local-airflow-scheduler-1", "airflow", "dags", "trigger", dag_id, "--run-id", run_id],
        capture_output=True, text=True, check=True,
    )
print(f"triggered {dag_id!r} as {run_id!r}")

deadline = time.time() + 90
state = "queued"
while state not in ("success", "failed") and time.time() < deadline:
    time.sleep(3)
    r = subprocess.run(
        ["docker", "exec", "airflow-local-airflow-scheduler-1", "airflow", "dags", "list-runs", "-d", dag_id, "--output", "json"],
        capture_output=True, text=True, check=True,
    )
    runs = json.loads(r.stdout)
    state = next(x["state"] for x in runs if x["run_id"] == run_id)
print(f"run {run_id!r} reached state: {state}")
assert state == "success", f"expected success, got {state}"

# Independent check: read the real file the task wrote, bypassing Airflow entirely.
import pandas as pd

raw = pd.read_csv(".airflow-local/dags/data/work/etl_ingest_demo/raw_trips.csv")
print(f"direct file check: {len(raw)} rows in raw_trips.csv")
print("columns:", list(raw.columns))
assert len(raw) == 81013, f"expected 81013, got {len(raw)}"

triggered 'cityflow_etl_ingest_demo' as 'lesson2_m5_verify_1784914105'
run 'lesson2_m5_verify_1784914105' reached state: success
direct file check: 81013 rows in raw_trips.csv
columns: ['tpep_pickup_datetime', 'tpep_dropoff_datetime', 'passenger_count', 'trip_distance', 'PULocationID', 'DOLocationID', 'payment_type', 'fare_amount', 'tip_amount', 'total_amount']

The task’s own log reports the same number from inside the container:

[2026-07-24T17:28:34.4Z] {logging_mixin.py:190} INFO - ingested 81013 real CityFlow trips (2024-01-01 to 2024-01-02, exclusive) -> /opt/airflow/dags/data/work/etl_ingest_demo/raw_trips.csv

Both checks agree: 81,013 real trips, because that’s genuinely how many real CityFlow rows picked up on 2024-01-01.

Key Takeaways

Real ingest means a real date-range filter on a real parsed datetime64 column against the actual parquet file — not a hardcoded sample CSV.
Ingest writes its result to a real file path, not a return value — XCom carries small facts (like a row count) between tasks, not the data itself once the data is real-sized.
81,013 real rows for one real day is already over 150x Module 4’s 500-row guided project — and Lesson 5 scales this same task to three real days without changing a line of its logic.
Two independent reads (the task’s own log, and this script’s direct pandas.read_csv of the file the task wrote) agreeing on the same count is the same “trust but verify” pattern every prior module used for hooks, sensors, and loads.

Exercises

Exercise 1 — Change INGEST_END_EXCLUSIVE to "2024-01-08" (a full real week) and re-trigger. Read the real row count from the task’s log — how many real CityFlow trips picked up in that first week of January 2024?

Exercise 2 — Add a second XCom-returned value, the real min and max tpep_pickup_datetime in the sliced data, and print them in the task’s log. Confirm they fall inside [INGEST_START, INGEST_END_EXCLUSIVE) as expected.

Exercise 3 — KEEP_COLUMNS drops VendorID and RatecodeID from the real source file. Add VendorID back to the list, re-trigger, and confirm raw_trips.csv now has 11 columns instead of 10 by reading the file directly, the same way this lesson’s verification script did.

Previous lesson

Lesson 1 - Designing the DAG

Next lesson

Lesson 3 - Validate and Transform Tasks

Courses

DATATWEETS

Title here

Lesson 2 - The Ingest Task

Welcome to the Ingest Task

CityFlow’s Real January Data

The Real Ingest Task

Triggered for Real: 81,013 Rows

Key Takeaways

Exercises

Lesson 2 - The Ingest Task

Welcome to the Ingest Task#

CityFlow’s Real January Data#

The Real Ingest Task#

Triggered for Real: 81,013 Rows#

Key Takeaways#

Exercises#

Welcome to the Ingest Task

CityFlow’s Real January Data

The Real Ingest Task

Triggered for Real: 81,013 Rows

Key Takeaways

Exercises