Data Pipelines

What a data pipeline is

A data pipeline is an automated process that takes data from one or more source systems, moves it to a store built for analysis, and turns it into tables, files or reports that people can use. It runs without anyone clicking, on a schedule or when new data arrives, and it is code: versioned, reviewed, tested and logged like any other program.

At Kestrel Energy the monthly key performance indicator (KPI) pack for Margriet de Wit (CFO) is still built by hand. Lucía Romero opens the nightly export of KBS (the Kestrel Billing System) in a spreadsheet, looks up tariffs, builds a pivot table and pastes the result into last month's workbook. Every step is a pipeline step, just done by a person: extract (the export), load (open it), transform (clean, look up, aggregate) and publish (paste into the pack). Joost Bakker, who owns KBS, is replacing it with an automated pipeline under change CHG-0772, "Nightly analytics load after the 02:30 export".

Two rules from the rest of this encyclopedia shape the design. The pipeline does not query the production database db01 during the day: Hamid Rahimi (database administrator) wants analytics kept off it, for the reasons in Why analytics doesn't run on the production database. And it reads the export KBS already produces at 02:30, instead of adding a second heavy job to the night.

ETL and ELT

The classic order is extract, transform, load: ETL cleans and reshapes data in a separate engine before it reaches the warehouse. Extract, load, transform (ELT) loads the data first, unchanged, and transforms it afterward inside the analytical store, usually with SQL. Most new pipelines are ELT, for practical reasons:

• The raw data is kept. When a transformation turns out to be wrong (a tariff mapped to the wrong group), you fix the SQL and rebuild from the raw copy. With ETL the original is gone, and you have to extract it again from a source that may have changed since.
• One engine does the work. Analytical databases and engines such as DuckDB are fast at exactly the joins and aggregations that transformations need.
• Transformations are plain SQL files that analysts can read and review, often managed with a tool such as dbt (see Warehouses, Lakes and Lakehouses).

ETL still has its place: when personal data must be removed or pseudonymized before it is stored anywhere else (see Data Governance and Privacy), or when the target cannot run heavy transformations itself. Many real pipelines are a mix: a light cleanup on the way in, the business logic afterward.

Batch and stream processing

Batch processing handles data in chunks at intervals: every night, every hour, the data of the previous period is processed in one run. A batch has a clear start and end, so it is easy to check ("did yesterday load completely?") and easy to repeat.

Stream processing handles each record or event continuously, seconds after it happens. The events usually travel through a message broker such as Apache Kafka, and a stream processor (Kafka Streams, Apache Flink, or a cloud service) keeps running aggregates. Kestrel's meter-api already publishes a ReadingReceived event per reading (see Event-Driven Architecture), so a stream is technically possible.

Choose by the decision the data supports, not by fashion. Margriet reads the KPI pack once a month; a nightly batch is more than fresh enough, and much simpler to run and to check. Many "real-time" requirements turn out to be "by the start of the working day".

Full and incremental loads

A full load copies everything every time. For Kestrel's 12-row customer table that is perfect: simple, and always right. For readings it is not: about 400,000 smart meters with a reading every 15 minutes produce 38.4 million rows a day (see Scalability and Caching Patterns). An incremental load processes only the data that is new or changed since the previous run.

To know what is new, the pipeline keeps a high-water mark: the highest value it has processed, saved at the end of each successful run. The question is which column. The obvious choice, the reading date, is wrong: a meter that was offline for three days sends three days at once, and their reading dates are already behind the mark. Use the moment the data arrived in the source, then reload every business day that the new rows touch.

# Readings as they sit in the source: (meter, reading day, arrived at, UTC)
source = [
    ("M-30001", "2026-10-04", "2026-10-05T00:41"),
    ("M-30002", "2026-10-04", "2026-10-05T00:43"),
    ("M-30001", "2026-10-05", "2026-10-06T00:40"),
    ("M-30002", "2026-10-05", "2026-10-06T00:44"),
    # M-30010 was offline and sent three days at once
    ("M-30010", "2026-10-02", "2026-10-05T16:12"),
    ("M-30010", "2026-10-03", "2026-10-05T16:12"),
    ("M-30010", "2026-10-04", "2026-10-05T16:12"),
    ("M-30010", "2026-10-05", "2026-10-06T00:45"),
]

# Saved by the previous run (ISO 8601 strings compare correctly as text)
last_run = {"max_reading_day": "2026-10-04", "max_arrived_at": "2026-10-05T02:00"}

# Wrong: a high-water mark on the business date misses late data
by_day = sorted({day for _, day, _ in source if day > last_run["max_reading_day"]})
print("watermark on reading day reloads:", by_day)

# Right: a high-water mark on arrival time, then reload every day it touches
new_rows = [row for row in source if row[2] > last_run["max_arrived_at"]]
by_arrival = sorted({day for _, day, _ in new_rows})
print("watermark on arrival reloads:   ", by_arrival)
print("next watermark:", max(arrived for _, _, arrived in new_rows))

watermark on reading day reloads: ['2026-10-05']
watermark on arrival reloads:    ['2026-10-02', '2026-10-03', '2026-10-04', '2026-10-05']
next watermark: 2026-10-06T00:45

The first approach silently loses three days of one meter: no error, just a lower total for those days. The second reloads four days. That is only safe because each day is reloaded as a whole, which is the subject of the next section.

Change data capture

A high-water mark needs a reliable "changed at" column, and it never sees deleted rows. Change data capture (CDC) reads the changes from the source database's own transaction log instead (the write-ahead log in PostgreSQL), and turns every insert, update and delete into an event, in commit order. Debezium describes itself as "an open source distributed platform for change data capture" and usually publishes those events to Kafka. Cloud data platforms offer CDC connectors as managed services.

CDC puts almost no query load on the source, catches deletes, and delivers changes within seconds. The costs: someone has to run the CDC platform, it needs elevated access to the database log (Hamid would have to grant replication rights on db01), and a schema change in the source flows straight into the pipeline. For an application that already publishes events through a transactional outbox, consuming those events is often the cleaner option: they carry business meaning ("reading received") rather than raw row changes.

Idempotent loads: safe to run twice

Pipelines fail halfway, time out and get retried, and people rerun them by hand. So every load must be idempotent: running it twice leaves the same result as running it once, the same idea as an idempotent script or an idempotent consumer. A plain INSERT is not. The examples load 12 September from Kestrel's sample readings (see Data Types and Data Quality for the data and its planted problems) into an analytics table, in DuckDB 1.5:

-- The analytics copy of the readings: the target of the nightly load
CREATE TABLE fact_reading (
  meter_id     VARCHAR,
  reading_date DATE,
  kwh_import   DECIMAL(8,3)
);

-- Naive load of 12 September, run twice (say, a retry after a timeout)
INSERT INTO fact_reading
SELECT meter_id, reading_date, kwh_import FROM daily_readings
WHERE reading_date = DATE '2026-09-12';
INSERT INTO fact_reading
SELECT meter_id, reading_date, kwh_import FROM daily_readings
WHERE reading_date = DATE '2026-09-12';

SELECT count(*) AS rows, sum(kwh_import) AS kwh FROM fact_reading;

┌───────┬───────────────┐
│ rows  │      kwh      │
│ int64 │ decimal(38,3) │
├───────┼───────────────┤
│    26 │       272.814 │
└───────┴───────────────┘

Twelve meters, yet 26 rows: the retry doubled the day, and the source already contained one duplicate (meter M-30005 appears twice on 12 September). That duplicate in the sample data is exactly what a non-idempotent load leaves behind. Nothing failed, and the dashboard would simply show twice the energy for that day.

The standard fix for a batch pipeline is to replace the whole partition: delete the day, then insert it again, in one transaction so nobody ever sees the day half-loaded. SELECT DISTINCT removes exact duplicates from the source on the way in.

CREATE TABLE fact_reading (
  meter_id     VARCHAR,
  reading_date DATE,
  kwh_import   DECIMAL(8,3)
);

-- Idempotent load of one day: replace the whole partition in one transaction
BEGIN TRANSACTION;
DELETE FROM fact_reading WHERE reading_date = DATE '2026-09-12';
INSERT INTO fact_reading
SELECT DISTINCT meter_id, reading_date, kwh_import FROM daily_readings
WHERE reading_date = DATE '2026-09-12';
COMMIT;

-- The retry: exactly the same statements again
BEGIN TRANSACTION;
DELETE FROM fact_reading WHERE reading_date = DATE '2026-09-12';
INSERT INTO fact_reading
SELECT DISTINCT meter_id, reading_date, kwh_import FROM daily_readings
WHERE reading_date = DATE '2026-09-12';
COMMIT;

SELECT count(*) AS rows, sum(kwh_import) AS kwh FROM fact_reading;

┌───────┬───────────────┐
│ rows  │      kwh      │
│ int64 │ decimal(38,3) │
├───────┼───────────────┤
│    12 │       128.100 │
└───────┴───────────────┘

The second option is an upsert on the business key: give the table a primary key on (meter, day) and let an existing row be replaced. The key is also a safety net: a careless plain INSERT now fails loudly instead of doubling quietly.

CREATE TABLE fact_reading (
  meter_id     VARCHAR,
  reading_date DATE,
  kwh_import   DECIMAL(8,3),
  PRIMARY KEY (meter_id, reading_date)
);

-- Upsert: a row with an existing key replaces the old one
INSERT OR REPLACE INTO fact_reading
SELECT DISTINCT meter_id, reading_date, kwh_import FROM daily_readings
WHERE reading_date = DATE '2026-09-12';
INSERT OR REPLACE INTO fact_reading
SELECT DISTINCT meter_id, reading_date, kwh_import FROM daily_readings
WHERE reading_date = DATE '2026-09-12';

SELECT count(*) AS rows, sum(kwh_import) AS kwh FROM fact_reading;

-- A naive INSERT of the same day is now refused instead of doubling
INSERT INTO fact_reading
SELECT meter_id, reading_date, kwh_import FROM daily_readings
WHERE reading_date = DATE '2026-09-12';

┌───────┬───────────────┐
│ rows  │      kwh      │
│ int64 │ decimal(38,3) │
├───────┼───────────────┤
│    12 │       128.100 │
└───────┴───────────────┘
Constraint Error: Duplicate key "meter_id: M-30001, reading_date: 2026-09-12" violates primary key constraint.

Other databases spell the upsert differently: MERGE in SQL Server, Oracle and recent PostgreSQL, or INSERT … ON CONFLICT … DO UPDATE in PostgreSQL and SQLite. With files instead of tables, the same idea is "overwrite the folder for that day" (see Warehouses, Lakes and Lakehouses).

Backfills

A backfill is a run of the pipeline over past periods: to fill a new table with history, or to repair data after a bug fix. Suppose that in November the team discovers that negative readings were loaded as zero all of September. After the fix, the backfill is simply the normal daily load, run once for each of the 30 September days. Because every day is replaced as a whole, the backfill cannot double anything, and it can be stopped and restarted halfway.

That is the real payoff of idempotent, per-day loads: repairs become routine instead of a special project with hand-written fix-up scripts. Run backfills in a quiet period and limit how many days run in parallel, so they do not crowd out the regular nightly run.

Orchestration: more than cron

The first version of many pipelines is a few cron entries: the export at 02:30, the load at 04:00, the report at 05:00. It works until the export runs late. The billing export has a service level objective of finishing by 06:00 (see Kestrel's SLOs), so on a busy night the 04:00 load reads yesterday's file, or half of today's, and the report happily publishes it. Cron knows times, not dependencies.

A workflow orchestrator is a scheduler that knows the steps of a pipeline and the dependencies between them. It starts a step only when the steps it depends on succeeded, retries failures with a delay, records every run, runs backfills for a date range, and alerts when something fails or runs too long. Pipelines are described as a directed acyclic graph (DAG) of tasks: arrows point from a step to the steps that need its output, and there are no cycles.

This is roughly what Kestrel's nightly job could look like in Airflow 3 (not run: Airflow is not installed here). Each function is a task; calling one with the result of another creates the dependency. The first task waits for the export instead of trusting the clock: if the export's completion marker is not there yet, it fails and Airflow retries it every 15 minutes.

from datetime import datetime, timedelta
from pathlib import Path
from zoneinfo import ZoneInfo

import pendulum
from airflow.sdk import dag, task

EXPORTS = Path("/srv/exports/kbs")
AMSTERDAM = ZoneInfo("Europe/Amsterdam")


def business_day(logical_date: datetime) -> str:
    """The day the data belongs to: the day before the run, Amsterdam time."""
    return (logical_date.astimezone(AMSTERDAM).date() - timedelta(days=1)).isoformat()


@dag(
    schedule="0 4 * * *",  # 04:00 every day
    start_date=pendulum.datetime(2026, 10, 1, tz="Europe/Amsterdam"),
    catchup=False,
    tags=["kpi-pack", "CHG-0772"],
)
def nightly_analytics_load():
    @task(retries=8, retry_delay=timedelta(minutes=15))
    def wait_for_export(logical_date=None):
        day = business_day(logical_date)
        marker = EXPORTS / day / "_SUCCESS"  # written last by the export job
        if not marker.exists():
            raise RuntimeError(f"export for {day} not finished yet")
        return str(marker.parent)

    @task
    def load_bronze(folder: str, logical_date=None):
        ...  # replace the bronze partition for business_day(logical_date)

    @task
    def build_gold(logical_date=None):
        ...  # run the SQL transformations for that day into a staging copy

    @task
    def check_quality(logical_date=None):
        ...  # row counts, duplicates, negative readings: fail on errors

    @task
    def publish(logical_date=None):
        ...  # swap the checked tables in; only reached when the checks pass

    load_bronze(wait_for_export()) >> build_gold() >> check_quality() >> publish()


nightly_analytics_load()

Airflow passes every task the run's logical_date, so the same code loads any day: that is what makes a backfill a matter of picking a date range. Mind which day that is. In Airflow 3, a cron schedule gets the logical date of the moment it runs (6 October, 04:00), not the start of the previous interval as in Airflow 2. So the tasks work out the business day, the day before, themselves. Material written for Airflow 2 is often out of date: in Airflow 3, execution_date is gone (use logical_date), Datasets were renamed Assets, SubDAGs were removed, and the TaskFlow decorators are imported from airflow.sdk.

Monitoring freshness and quality

A failing web server produces errors; a pipeline that did not run produces nothing at all. The dashboard keeps showing yesterday's numbers, and nobody notices until someone asks why the September figures look like August. Silence is not success.

Data freshness is how recent the data in an analytical store is: the time since the newest data was successfully loaded, or the newest business day it covers. It is a natural service level indicator: "the readings were loaded less than 26 hours ago" (a day plus some slack). The pipeline writes a row to a log table after every successful run, and an independent check reads that table:

-- Written by the pipeline: one row per successful load run
CREATE TABLE load_log (
  dataset     VARCHAR,
  loaded_for  DATE,        -- the business day the run loaded
  finished_at TIMESTAMP    -- when the run finished, in UTC
);
INSERT INTO load_log VALUES
  ('readings', DATE '2026-10-04', TIMESTAMP '2026-10-05 02:12:40'),
  ('readings', DATE '2026-10-05', TIMESTAMP '2026-10-06 02:09:15'),
  ('invoices', DATE '2026-10-03', TIMESTAMP '2026-10-04 02:30:55');

-- The check at 07:00 Amsterdam time on 6 October (05:00 UTC)
SELECT dataset,
       max(loaded_for)                                    AS newest_day,
       TIMESTAMP '2026-10-06 05:00:00' - max(finished_at) AS age,
       CASE WHEN TIMESTAMP '2026-10-06 05:00:00' - max(finished_at)
                 > INTERVAL 26 HOUR
            THEN 'STALE' ELSE 'ok' END                    AS status
FROM load_log
GROUP BY dataset
ORDER BY dataset;

┌──────────┬────────────┬─────────────────┬─────────┐
│ dataset  │ newest_day │       age       │ status  │
│ varchar  │    date    │    interval     │ varchar │
├──────────┼────────────┼─────────────────┼─────────┤
│ invoices │ 2026-10-03 │ 2 days 02:29:05 │ STALE   │
│ readings │ 2026-10-05 │ 02:50:45        │ ok      │
└──────────┴────────────┴─────────────────┴─────────┘

The invoice load has not succeeded for two nights, and no job reported an error: perhaps it was never scheduled after a server move. Only a check that looks for the presence of fresh data catches that. Run the check outside the pipeline itself (a pipeline that does not run cannot report that it did not run), page or ticket on STALE, and show the newest_day as "data as of" on the dashboard, so readers can see it too. The same idea in monitoring is the "watchdog" alert described in Alerting.

Freshness is one check among several. Before publishing, a pipeline compares what it loaded with what it expected:

• Volume: about 12 readings per day in the sample, about 38.4 million in production. A day with half that is wrong even if every row is valid.
• Uniqueness and keys: no duplicate (meter, day), no contract without a customer.
• Ranges: no negative imports, no 10-fold jumps without a flag.
• Reconciliation: the total billed in the KPI tables equals the total in the KBS export.

A failed check stops the publish step, so the dashboard keeps yesterday's correct figures, clearly dated, instead of today's wrong ones. How to write such tests as part of the transformation code is shown in Warehouses, Lakes and Lakehouses.

Time zones and daylight saving time

"One day of readings" sounds like a fixed amount. It is not. Smart meters measure every 15 minutes, so a normal day has 96 readings, but on the days that clocks change in the Netherlands a local calendar day is 23 or 25 hours long. Python's zoneinfo module knows the rules:

from datetime import datetime, timedelta
from zoneinfo import ZoneInfo

AMSTERDAM = ZoneInfo("Europe/Amsterdam")
UTC = ZoneInfo("UTC")


def quarter_hours(year, month, day):
    """Number of 15-minute readings in one local calendar day."""
    start = datetime(year, month, day, tzinfo=AMSTERDAM)
    end = start + timedelta(days=1)  # same wall-clock time, next day
    seconds = end.astimezone(UTC) - start.astimezone(UTC)
    return int(seconds.total_seconds() // 900)


for day in [(2026, 3, 29), (2026, 10, 6), (2026, 10, 25)]:
    print(datetime(*day).date(), quarter_hours(*day), "quarter-hours")

2026-03-29 92 quarter-hours
2026-10-06 96 quarter-hours
2026-10-25 100 quarter-hours

A volume check that expects 96 readings per meter will flag every meter on Sunday 25 October 2026. Worse, a pipeline that stores local times without an offset has two readings for 02:15 that night and cannot tell them apart. The robust pattern:

• Store timestamps in Coordinated Universal Time (UTC), or with their offset (ISO 8601: 2026-10-25T02:15:00+02:00).
• Convert to local time only to decide which business day a reading belongs to, and in reports.
• Compute the expected volume per day from the calendar, as above, instead of hard-coding 96.
• Schedule jobs with an explicit time zone. "04:00" in a crontab means the server's time zone, which may be UTC on one machine and Amsterdam time on the next.

A pipeline checklist

What separates a pipeline that can be trusted from a script that usually works:

• Code in version control, deployed through a change (CHG-0772 at Kestrel), not edited on the server.
• Idempotent per partition, so any run, retry or backfill can be repeated.
• Incremental by arrival, with overlap, so late data is picked up.
• Dependencies, not clock times, between steps.
• Quality checks that stop publishing, and a freshness check that runs independently.
• Logs per run (what day, how many rows, how long), kept as described in Scheduling and Logging.
• An owner who gets the alert, and a short runbook for the usual failures.
• Personal data handled on purpose: only the columns the analysis needs, and retention applied (see Data Governance and Privacy).