Data Engineering: Week 2b: Data Ingestion with Airflow

The first Week 2 article introduced data lakes, warehouses, processing, and orchestration. This article turns those ideas into a small Airflow pipeline. It uses the current Airflow Docker Compose quick-start, stores intermediate data in Google Cloud Storage so any worker can reach it, and loads a Parquet object into BigQuery.

The pipeline

The example has four stages:

download the public NYC taxi Parquet file into a GCS bucket;
validate its required columns, normalize its timestamps, and write a typed Parquet object;
load the Parquet object into a managed BigQuery table;
inspect task state and logs in the Airflow UI.

The task functions run independently of the DAG parser. A task may run on another worker or at a later time, so it must read its inputs from a durable shared system. A local path on one worker is not a safe handoff. The example uses GCS for this reason. If a local-only experiment uses a file volume instead, mount the same volume into every worker and document that constraint.

A single local installation

Use the Airflow 3.3.0 Docker Compose file throughout this walkthrough. It includes PostgreSQL for Airflow metadata, Redis as the Celery broker, the API server, scheduler, workers, triggerer, and an initialization service. Docker Compose v2.14 or newer and at least 4 GB of memory are recommended by the Airflow Docker guide.

With Python 3 available on the host, create a project directory and the directories mounted by the official file. Generate the encryption key once and retain the same .env file when restarting this environment:

mkdir -p airflow-example/{dags,logs,plugins,config,data}
cd airflow-example
curl -LfO 'https://airflow.apache.org/docs/apache-airflow/3.3.0/docker-compose.yaml'
echo "AIRFLOW_UID=$(id -u)" > .env
echo "FERNET_KEY=$(python3 -c 'import base64, secrets; print(base64.urlsafe_b64encode(secrets.token_bytes(32)).decode())')" >> .env

The Google provider and the small example need a few additional packages. For a learning environment, the official image can install them at startup through _PIP_ADDITIONAL_REQUIREMENTS:

echo '_PIP_ADDITIONAL_REQUIREMENTS=apache-airflow-providers-google==22.4.0 pandas==2.2.3 pyarrow==18.1.0' >> .env
echo 'AIRFLOW_CONN_GOOGLE_CLOUD_DEFAULT=google-cloud-platform://' >> .env

The DAG tasks run inside Airflow containers, so host environment variables and host ADC files must be forwarded explicitly. Set the resource variables first, then create an override file. This mounts the host’s read-only ADC file into every Airflow process that parses, triggers, or executes the DAG:

gcloud auth application-default login

export GCP_PROJECT_ID="YOUR_GCP_PROJECT"
export GCP_REGION="europe-west1"
export INGESTION_BUCKET="YOUR_UNIQUE_BUCKET"
export BQ_DATASET="nytaxi"
export GOOGLE_APPLICATION_CREDENTIALS="${GOOGLE_APPLICATION_CREDENTIALS:-$HOME/.config/gcloud/application_default_credentials.json}"
test -r "$GOOGLE_APPLICATION_CREDENTIALS"

cat > docker-compose.override.yml <<'YAML'
x-google-auth: &google-auth
  environment:
    GOOGLE_APPLICATION_CREDENTIALS: /opt/airflow/credentials/application_default_credentials.json
    GCP_PROJECT_ID: ${GCP_PROJECT_ID}
    GCP_REGION: ${GCP_REGION}
    INGESTION_BUCKET: ${INGESTION_BUCKET}
    BQ_DATASET: ${BQ_DATASET}
  volumes:
    - ${GOOGLE_APPLICATION_CREDENTIALS}:/opt/airflow/credentials/application_default_credentials.json:ro

services:
  airflow-apiserver: *google-auth
  airflow-scheduler: *google-auth
  airflow-dag-processor: *google-auth
  airflow-worker: *google-auth
  airflow-triggerer: *google-auth
  airflow-cli: *google-auth
YAML

For a long-lived deployment, build a custom image with these requirements under the Airflow constraints for the chosen Python version. Installing packages at every container start is convenient for a tutorial, but it makes startup slower and is less reproducible.

Initialize and start the services:

docker compose up airflow-init
docker compose up -d
docker compose ps

The expected result is an initialized metadata database and healthy Airflow services. Open http://localhost:8080 after the initialization service completes. If a service is not healthy, inspect the relevant service:

docker compose ps
docker compose logs --tail=100 airflow-scheduler
docker compose run --rm airflow-cli airflow dags list-import-errors

The official compose file mounts dags, logs, plugins, and config into each Airflow service. Put the DAG below in dags/nyc_taxi_gcs.py. The data itself stays in GCS, so it does not depend on one worker’s local filesystem.

Authentication and variables

The local setup uses the Google Cloud CLI to create Application Default Credentials and mounts that file read-only. In a managed Composer environment, use the environment’s runtime service account instead of the local credential mount. Keep credentials outside the repository.

The bucket must already exist, and the identity used by Airflow needs permission to create objects in it and load BigQuery tables.

docker compose config

The override forwards the project, bucket, dataset, and credential settings to the Airflow services. These values must agree with the resources and permissions in the target project.

The GCS and BigQuery DAG

Airflow 3 exposes DAG authoring through the stable airflow.sdk namespace. The schedule uses a timezone-aware pendulum start date. The Google provider uses the documented GCSToBigQueryOperator interface; this example uses the storage client inside tasks so the download and transformation can both write durable objects.

"""Download, validate, transform, and load one NYC taxi file."""

from __future__ import annotations

import io
import os
import tempfile
import urllib.request
from datetime import timedelta

import pandas as pd
import pendulum
from google.cloud import storage

from airflow.providers.google.cloud.transfers.gcs_to_bigquery import (
    GCSToBigQueryOperator,
)
from airflow.sdk import dag, task


PROJECT_ID = os.environ.get("GCP_PROJECT_ID", "YOUR_GCP_PROJECT")
BUCKET = os.environ.get("INGESTION_BUCKET", "YOUR_UNIQUE_BUCKET")
DATASET = os.environ.get("BQ_DATASET", "nytaxi")
RAW_OBJECT = "raw/yellow_tripdata_2021-01.parquet"
PROCESSED_OBJECT = "processed/yellow_tripdata_2021-01.parquet"
SOURCE_URL = (
    "https://d37ci6vzurychx.cloudfront.net/trip-data/"
    "yellow_tripdata_2021-01.parquet"
)
GCP_CONN_ID = "google_cloud_default"


@dag(
    dag_id="nyc_taxi_gcs_to_bigquery",
    schedule="@monthly",
    start_date=pendulum.datetime(2024, 1, 1, tz="UTC"),
    catchup=False,
    tags=["example", "gcs", "bigquery"],
)
def nyc_taxi_pipeline():
    @task(retries=2, retry_delay=timedelta(minutes=5))
    def download_to_gcs() -> str:
        request = urllib.request.Request(
            SOURCE_URL,
            headers={"User-Agent": "airflow-ingestion-example"},
        )
        client = storage.Client(project=PROJECT_ID)
        blob = client.bucket(BUCKET).blob(RAW_OBJECT)
        with urllib.request.urlopen(request, timeout=120) as response:
            blob.upload_from_file(
                response,
                content_type="application/octet-stream",
            )
        return RAW_OBJECT

    @task
    def validate_and_convert(raw_object: str) -> str:
        client = storage.Client(project=PROJECT_ID)
        source = client.bucket(BUCKET).blob(raw_object)
        payload = source.download_as_bytes()
        frame = pd.read_parquet(io.BytesIO(payload))
        required = {"tpep_pickup_datetime", "tpep_dropoff_datetime"}
        missing = required.difference(frame.columns)
        if missing:
            raise ValueError(f"missing required columns: {sorted(missing)}")

        for column in required:
            frame[column] = pd.to_datetime(frame[column], errors="raise")

        with tempfile.NamedTemporaryFile(suffix=".parquet") as output:
            frame.to_parquet(output.name, index=False)
            client.bucket(BUCKET).blob(PROCESSED_OBJECT).upload_from_filename(
                output.name,
                content_type="application/octet-stream",
            )
        return PROCESSED_OBJECT

    raw_object = download_to_gcs()
    processed_object = validate_and_convert(raw_object)

    load_to_bigquery = GCSToBigQueryOperator(
        task_id="load_to_bigquery",
        bucket=BUCKET,
        source_objects=[processed_object],
        destination_project_dataset_table=(
            f"{PROJECT_ID}.{DATASET}.yellow_tripdata"
        ),
        source_format="PARQUET",
        write_disposition="WRITE_TRUNCATE",
        create_disposition="CREATE_IF_NEEDED",
        autodetect=True,
        project_id=PROJECT_ID,
        location=os.environ.get("GCP_REGION", "europe-west1"),
        gcp_conn_id=GCP_CONN_ID,
    )

    processed_object >> load_to_bigquery


nyc_taxi_pipeline()

The task return values are short object names. Airflow can store those values as XComs, while the CSV and Parquet payloads remain in GCS. The transformation above reads one file into memory for clarity; for a much larger file, use a distributed Beam or Spark transformation and keep the same durable-object boundary.

The operator’s source_objects argument accepts templated object names, and destination_project_dataset_table uses the documented project, dataset, and table form. The google_cloud_default connection supplies credentials. In Composer, the runtime identity is normally attached to the environment. Locally, ADC supplies credentials to the Google client and provider hook.

Inspecting a run

After placing the DAG in dags/, refresh the UI and trigger nyc_taxi_gcs_to_bigquery. The expected sequence is download_to_gcs, validate_and_convert, and load_to_bigquery, with the downstream task waiting for the processed object. The tree and graph views make this dependency visible.

From the CLI, the same checks are available with the current Airflow commands:

docker compose run --rm airflow-cli airflow dags list
docker compose run --rm airflow-cli airflow dags show nyc_taxi_gcs_to_bigquery
docker compose logs --tail=100 airflow-worker

Local Postgres and pgAdmin

The Airflow metadata database is an internal service of the official compose project. For a separate local data store, save the following as local-data-services.yml. It maps Postgres to port 5433 and pgAdmin to 8081, so pgAdmin does not collide with the Airflow web interface on port 8080.

services:
  data-postgres:
    image: postgres:16
    environment:
      POSTGRES_USER: taxi
      POSTGRES_PASSWORD: ${POSTGRES_PASSWORD:?set POSTGRES_PASSWORD}
      POSTGRES_DB: ny_taxi
    ports:
      - "5433:5432"
    volumes:
      - data-postgres:/var/lib/postgresql/data

  pgadmin:
    image: dpage/pgadmin4:9
    environment:
      PGADMIN_DEFAULT_EMAIL: ${PGADMIN_EMAIL:?set PGADMIN_EMAIL}
      PGADMIN_DEFAULT_PASSWORD: ${PGADMIN_PASSWORD:?set PGADMIN_PASSWORD}
    ports:
      - "8081:80"
    depends_on:
      - data-postgres

volumes:
  data-postgres:

Set the three values and start the services:

export POSTGRES_PASSWORD="choose-a-local-password"
export PGADMIN_EMAIL="you@example.com"
export PGADMIN_PASSWORD="choose-an-admin-password"
docker compose -f local-data-services.yml up -d

Postgres is the database service. pgAdmin is a separate administration interface that connects to it. In pgAdmin, add a server with host data-postgres, port 5432, database ny_taxi, and the credentials from the Compose environment. From the host, use localhost:5433; container-to-container connections use the service name and the container port.

For a small local loading script, use the archived January 2021 CSV release and iterate over its compressed chunks directly. The loop terminates when the iterator is exhausted, and the connection details come from an environment variable:

import os

import pandas as pd
from sqlalchemy import create_engine


engine = create_engine(os.environ["POSTGRES_URL"])
source_url = (
    "https://github.com/DataTalksClub/nyc-tlc-data/releases/download/"
    "yellow/yellow_tripdata_2021-01.csv.gz"
)

for chunk in pd.read_csv(
    source_url,
    compression="gzip",
    chunksize=100_000,
):
    chunk.to_sql("yellow_tripdata", engine, if_exists="append", index=False)

Set POSTGRES_URL to a URL such as postgresql+psycopg2://taxi:@localhost:5433/ny_taxi for a host process. A task running in another container must use a hostname reachable from that container, and a downstream Airflow worker should read a durable object or database table instead of assuming a file on an upstream worker.

Google Cloud Storage Transfer Service

Airflow is useful when a transfer is one step in a larger workflow. Storage Transfer Service is a managed option when the job is primarily copying objects from supported cloud or on-premises sources into Cloud Storage. Its schedule, source credentials, destination bucket, and overwrite policy are configured as a transfer job.

Further Airflow deployment material

The local Compose setup is for learning. For a Kubernetes deployment, start with the official Airflow Helm chart documentation and design the metadata database, secrets, logging, workers, and upgrades explicitly.

Cleanup

The following commands remove local containers, volumes, and their stored data. Run them only when you intend to discard the local environment:

docker compose down --volumes --remove-orphans
docker compose -f local-data-services.yml down --volumes --remove-orphans

Cloud Storage objects and BigQuery tables are separate resources; delete them through their normal project-specific lifecycle process when they are no longer needed.