Building an ETL Pipeline on AWS Step by Step

An ETL pipeline is a short program with a long list of ways to go wrong: a file arrives twice, a run fails halfway, bad rows slip through, yesterday's rerun doubles the data. The AWS services involved are the easy part. What makes a pipeline dependable is its structure, and the structure is the same whichever service runs it.

This walkthrough builds that structure in six steps and runs it. The code uses boto3 against moto, a local emulator of S3 and the Glue Data Catalog, with the transformation done in-process. That keeps the whole pipeline visible in one script. Afterwards, each step is mapped to the AWS service that would run it in production.

The design in one paragraph

Raw files land under a month prefix in S3 and are never modified. A run takes a month as its parameter, reads every raw file for that month, rejects rows that break named rules and counts them, writes the clean rows as Parquet to that month's partition after clearing it, registers the partition in the catalog, and writes a report proving that every input row is accounted for. Running it again for the same month produces the same result.

The pipeline

import io, json, os
import boto3
import pyarrow as pa
import pyarrow.compute as pc
import pyarrow.parquet as pq
from moto import mock_aws

os.environ.update(AWS_ACCESS_KEY_ID="testing", AWS_SECRET_ACCESS_KEY="testing",
                  AWS_DEFAULT_REGION="us-east-1")
BUCKET = "demo-pipeline"
COLUMNS = ["tpep_pickup_datetime", "tpep_dropoff_datetime", "PULocationID", "DOLocationID",
           "trip_distance", "payment_type", "fare_amount"]


def run(s3, glue, month):
    """One pipeline run for one month. Safe to repeat: it replaces that month's output."""
    # Extract: every raw object that arrived for this month
    raw_prefix = f"raw/trips/month={month}/"
    keys = [o["Key"] for o in s3.list_objects_v2(Bucket=BUCKET, Prefix=raw_prefix).get("Contents", [])]
    tables = [pq.read_table(io.BytesIO(s3.get_object(Bucket=BUCKET, Key=k)["Body"].read()), columns=COLUMNS)
              for k in keys]
    trips = pa.concat_tables(tables)

    # Validate: named rules, each counted; a row is charged to the first rule it breaks
    rules = {
        "outside_month": pc.not_equal(pc.strftime(trips["tpep_pickup_datetime"], format="%Y-%m"), month),
        "ends_before_start": pc.less(trips["tpep_dropoff_datetime"], trips["tpep_pickup_datetime"]),
        "negative_fare": pc.less(trips["fare_amount"], 0),
        "zero_distance": pc.equal(trips["trip_distance"], 0),
    }
    rejected, bad = {}, pa.array([False] * trips.num_rows)
    for name, breaks in rules.items():
        first = pc.and_(breaks, pc.invert(bad))
        rejected[name] = pc.sum(first).as_py()
        bad = pc.or_(bad, breaks)
    clean = trips.filter(pc.invert(bad))

    # Load: remove the partition's old objects, then write the new one
    out_prefix = f"curated/trips/pickup_month={month}/"
    for old in s3.list_objects_v2(Bucket=BUCKET, Prefix=out_prefix).get("Contents", []):
        s3.delete_object(Bucket=BUCKET, Key=old["Key"])
    buffer = io.BytesIO()
    pq.write_table(clean, buffer, compression="zstd")
    s3.put_object(Bucket=BUCKET, Key=out_prefix + "part-0000.parquet", Body=buffer.getvalue())

    # Register the partition if the catalog does not have it yet
    location = f"s3://{BUCKET}/{out_prefix}"
    existing = [p["Values"] for p in glue.get_partitions(DatabaseName="lake", TableName="trips")["Partitions"]]
    if [month] not in existing:
        glue.create_partition(DatabaseName="lake", TableName="trips",
                              PartitionInput={"Values": [month], "StorageDescriptor": {"Location": location}})

    # Reconcile and record
    report = {"month": month, "raw_files": len(keys), "raw_rows": trips.num_rows,
              "rejected": rejected, "loaded_rows": clean.num_rows,
              "reconciles": trips.num_rows == clean.num_rows + sum(rejected.values())}
    s3.put_object(Bucket=BUCKET, Key=f"runs/trips/{month}.json", Body=json.dumps(report).encode())
    return report


@mock_aws
def main():
    s3, glue = boto3.client("s3"), boto3.client("glue")
    s3.create_bucket(Bucket=BUCKET)
    s3.upload_file("yellow_tripdata_2024-01.parquet", BUCKET, "raw/trips/month=2024-01/yellow_tripdata_2024-01.parquet")
    glue.create_database(DatabaseInput={"Name": "lake"})
    glue.create_table(DatabaseName="lake", TableInput={
        "Name": "trips", "PartitionKeys": [{"Name": "pickup_month", "Type": "string"}],
        "StorageDescriptor": {"Location": f"s3://{BUCKET}/curated/trips/", "Columns": []}})

    for attempt in (1, 2):
        report = run(s3, glue, "2024-01")
        objects = s3.list_objects_v2(Bucket=BUCKET, Prefix="curated/")["Contents"]
        partitions = glue.get_partitions(DatabaseName="lake", TableName="trips")["Partitions"]
        print(f"run {attempt}: raw {report['raw_rows']:,}  loaded {report['loaded_rows']:,}  "
              f"reconciles {report['reconciles']}  curated objects {len(objects)}  partitions {len(partitions)}")
    print("rejected by rule:", report["rejected"])


main()

run 1: raw 2,964,624  loaded 2,870,115  reconciles True  curated objects 1  partitions 1
run 2: raw 2,964,624  loaded 2,870,115  reconciles True  curated objects 1  partitions 1
rejected by rule: {'outside_month': 18, 'ends_before_start': 56, 'negative_fare': 37447, 'zero_distance': 56988}

Step 1: extract by parameter, not by clock

The run function takes month as an argument. It does not ask what today's date is. That one decision makes backfills, reruns and tests possible: any month can be processed at any time. It lists every object under that month's raw prefix, so a month delivered as several files is handled the same way as one.

Step 2: validate with named, counted rules

Four rules reject rows: pickup outside the month, a trip that ends before it starts, a negative fare, and zero distance. Each has a name and a count. A row that breaks several rules is charged to the first, so the counts add up.

The output shows what the data, one month of New York yellow taxi trips (NYC TLC: trip record data), actually contains: 18 rows outside the month, 56 ending before they start, 37,447 negative fares and 56,988 zero-distance trips. Whether zero-distance trips should be rejected is a business decision, and the count is what lets someone make it. A pipeline that silently drops 94,509 rows is hiding a decision; one that reports them by rule is documenting it.

In production, keep the rejected rows in a quarantine prefix instead of discarding them, so they can be inspected and reprocessed.

Step 3: transform

Here the transformation is only column selection and filtering. Real pipelines add type casting, derived columns, joins to reference data and deduplication. The principle that matters is to write it as a function of its inputs with no hidden state, so it can be tested with a small file.

Step 4: load by replacing the partition

The load clears the month's partition and writes it again. It never appends. This is what makes the second run harmless: both runs end with one object in the curated prefix and the same 2,870,115 rows.

Appending is the natural mistake. Glue's own documentation makes the responsibility explicit: when job bookmarks are not used, the job processes the entire dataset each run, and you are responsible for managing the output from previous runs (Glue docs: job bookmarks). Replacing a partition is the simplest way to meet that responsibility. Idempotency in data pipelines covers the alternatives, including merge.

One caution about the simple version shown: between the delete and the write, the partition is briefly empty. For tables read around the clock, write to a new location and switch the partition's location in the catalog, or use a table format with atomic commits such as Iceberg or Delta Lake.

Step 5: register in the catalog

Files in S3 are invisible to Athena and Redshift until the catalog knows about the partition. The run registers it if it is missing, and does nothing if it is already there, so the catalog step is repeatable too. Both runs end with one partition.

Step 6: reconcile and record

The last step checks that raw rows equal loaded rows plus rejected rows, and writes a small report to S3. reconciles True is the single most useful line a pipeline can print. A report per run also gives you history: when the rejection count for a rule jumps from tens to thousands, something changed upstream, and you can see when.

The same steps on AWS services

Three notes on that table:

• The engine depends on size. A month of this dataset fits in memory on one machine. At larger volumes the same logic moves into a Glue Spark job; Glue vs EMR vs Lambda covers the choice, and AWS Glue jobs with PySpark the script.
• SQL is an option. Athena's CTAS and INSERT INTO statements can partition, compress and convert data, and the tables and partitions they create are added to the Glue Data Catalog automatically (Athena docs: CTAS and INSERT INTO for ETL).
• Orchestration is separate from logic. Step Functions can control Glue to build ETL workflows (Step Functions docs: what is Step Functions), adding retries and failure notifications around steps that stay unchanged. See AWS Step Functions explained.

Before calling it production

• Permissions. One IAM role for the pipeline, limited to its prefixes and tables.
• Alerting. A failed run, or a report where reconciles is false, must reach a person. Glue publishes events, logs and metrics to CloudWatch for this (Glue docs: monitoring).
• Thresholds. Fail the run if rejections exceed an agreed share, instead of loading a suspiciously small table.
• Schema checks. Compare incoming columns and types with what the pipeline expects before transforming.
• Tests. Run the transform on a small fixture file in CI.
• Infrastructure as code. Define buckets, roles, jobs and schedules in a template, not by hand.
• Cost guards. Timeouts and worker counts on jobs; lifecycle rules on raw and temporary prefixes.

For the lake this pipeline writes into, see building a data lake on AWS with S3 and Glue; for the same discipline without any cloud at all, building your first data pipeline.

Building an end-to-end ETL pipeline is one of the projects in the Data Engineering course, which covers ETL and ELT design with Spark and AWS services.

Sources

• AWS Glue concepts (AWS Glue User Guide) — AWS Documentation
• Tracking processed data using job bookmarks (AWS Glue User Guide) — AWS Documentation
• Use CTAS and INSERT INTO for ETL and data analysis (Amazon Athena User Guide) — AWS Documentation
• What is Step Functions? (AWS Step Functions Developer Guide) — AWS Documentation
• Monitoring AWS Glue (AWS Glue User Guide) — AWS Documentation
• TLC Trip Record Data (NYC Taxi and Limousine Commission) — NYC Taxi and Limousine Commission