A First End-to-End Pipeline

Concepts covered: Batch Processing, Full vs Incremental Load

This section runs the shop's pipeline for 14 March from the source through to the consumer that reads the result, with 1 transform and 1 destination in between. Each step produces something that can be opened and counted, and knowing what each one produces is how a broken run is diagnosed in the morning.

Step 1: extract, 02:00 to 02:03

SELECT order_id, customer_email, category, amount, created_at
FROM orders
WHERE created_at >= '2024-03-14 00:00:00'
  AND created_at <  '2024-03-15 00:00:00';

The extract step reads 1 day's window from Postgres and writes the rows unchanged. It produces 6,214 rows in 1 Parquet file of 410 KB at s3://shop-raw/orders/date=2024-03-14/, and holds a connection to the checkout database for 3 minutes at the quietest hour of the day. Nothing is fixed here on purpose: amounts are still text, timestamps are still in the application's zone, and the developers' test orders are still present, because the raw copy is the one thing in the pipeline that matches what the source said that night. Postgres rows change afterwards as refunds and edits land.

Step 2: transform, 02:05 to 02:20

SELECT
  CAST(CONVERT_TIMEZONE('Europe/London', 'UTC', created_at) AS DATE) AS order_date,
  category,
  COUNT(*)                           AS orders,
  SUM(CAST(amount AS DECIMAL(10,2))) AS revenue
FROM raw_orders
WHERE customer_email NOT LIKE '%@kitchenco.com'
GROUP BY 1, 2;

The transform reads the 6,214 raw rows and produces 8: 1 per category for the day. On the way it casts amount to a decimal, drops 24 test orders, converts timestamps to UTC and sums what remains. The 6,190 real orders total 180,757.80 across 8 categories. An engineer compares the row count and the sum against the source the first time the chart looks wrong, so the step logs both.

4 of the 8 rows the transform produced; the other 4 categories (boards and utensils, storage and glassware) account for the remaining 2,046 orders.

Step 3: load, 02:20 to 02:22

The load step writes the 8 rows into the Snowflake table daily_category_revenue. What it produces is a new state of that table: 8 rows for 14 March alongside every earlier day. It deletes any rows already dated 14 March before inserting, so a rerun after a failure leaves 8 rows for the day rather than 16. A month holds 240 rows and a year 2,920. The run reports success at 02:22, and the on-call engineer sees 1 line in the morning: 3 steps, 6,214 rows in, 8 rows out, 22 minutes.

Step 4: the consumer reads

SELECT order_date, category, revenue
FROM daily_category_revenue
WHERE order_date >= DATEADD(day, -30, CURRENT_DATE)
ORDER BY order_date, category;

The chart on screen is produced by nothing the finance lead can see: a 3-minute extract, a 15-minute transform, a 2-minute load, all finished before 02:30, and a query over 240 rows when she opens the page. The freshness is what batch costs. The order placed at 00:05 on 14 March reached the chart on the morning of the 15th, and an order placed 5 minutes after the extract ran waits for the next night's run.

• 32 hours
• 5h 38m

On the morning of 15 March the chart shows revenue for 13 March and nothing for 14 March. daily_category_revenue holds no rows dated 14 March. The S3 folder date=2024-03-14/ holds a 410 KB Parquet file written at 02:03. Which step failed?

• The extract step, because the chart is missing a day
• The transform or the load, because raw rows exist and table rows do not
• The consumer, because Looker cached the previous day
• Postgres, because the orders were never recorded

About this lesson

This section is part of the Pipeline Anatomy: Beginner lesson on DataDriven, the home of data engineering. Lessons cover SQL, Python, Spark, data modeling and pipeline architecture, and each section pairs its explanation with worked examples and practice you run as you go.