ETL is the pattern of pulling data out of operational systems, reshaping it, and loading it into an analytical store. This page adds the architecture view, the ETL versus ELT decision, a runnable incremental pipeline, and the operational concerns that separate a demo script from a production job.
Most pipelines insert a staging area between extract and transform. Landing an unmodified copy of the source first means a bad transformation can be fixed and replayed without hitting the source system again, and it gives auditors a record of exactly what was received. The transform step applies cleaning, type conversion, deduplication, joins to reference data and aggregation. The load step writes into target tables, usually with a merge (upsert) so reruns do not create duplicates.
Classic ETL transforms data on a dedicated engine before it reaches the warehouse. ELT loads raw data first and transforms it with SQL inside the warehouse or lakehouse (Snowflake, BigQuery, Databricks), often orchestrated by dbt. ELT has become the default for cloud warehouses because compute scales independently and raw history stays queryable. ETL remains the right choice when data must be filtered or masked before it lands (personal data, regulated fields) or when the target cannot run heavy transformations.
import sqlite3
import pandas as pd
src = sqlite3.connect("source.db")
dwh = sqlite3.connect("warehouse.db")
dwh.execute("""CREATE TABLE IF NOT EXISTS fact_orders(
order_id INTEGER PRIMARY KEY, customer TEXT, amount REAL, updated_at TEXT)""")
dwh.execute("CREATE TABLE IF NOT EXISTS etl_watermark(name TEXT PRIMARY KEY, ts TEXT)")
# 1. Extract only rows changed since the last run (watermark)
row = dwh.execute("SELECT ts FROM etl_watermark WHERE name='orders'").fetchone()
since = row[0] if row else "1970-01-01"
df = pd.read_sql("SELECT * FROM orders WHERE updated_at > ?", src, params=[since])
# 2. Transform: clean and validate
df["customer"] = df["customer"].str.strip().str.lower()
df = df.dropna(subset=["order_id", "amount"])
assert (df["amount"] >= 0).all(), "negative amounts in batch"
# 3. Load idempotently (upsert) and advance the watermark in one transaction
with dwh:
dwh.executemany(
"""INSERT INTO fact_orders VALUES (?,?,?,?)
ON CONFLICT(order_id) DO UPDATE SET customer=excluded.customer,
amount=excluded.amount, updated_at=excluded.updated_at""",
df[["order_id", "customer", "amount", "updated_at"]].itertuples(index=False))
if len(df):
dwh.execute("INSERT OR REPLACE INTO etl_watermark VALUES ('orders', ?)",
[df["updated_at"].max()])
print(f"loaded {len(df)} rows")Three properties make this safe to schedule: it is incremental (a watermark limits extraction to changed rows), idempotent (an upsert means rerunning the same batch produces the same result), and atomic (the data and the watermark commit together, so a crash never skips rows).
| Approach | Latency | Typical tools | Use when |
|---|---|---|---|
| Nightly batch | Hours | Airflow, cron, SQL | Reporting, finance close |
| Micro-batch | Minutes | Spark Structured Streaming, Delta Live Tables | Near-real-time dashboards |
| Streaming / CDC | Seconds | Kafka, Debezium, Flink | Operational analytics, alerts |
In production, each stage runs as a task in an orchestrator such as Airflow, Dagster or Databricks Workflows, which handles scheduling, retries with backoff, dependencies between tables, and backfills over historical date ranges. Every run should record row counts in and out, rejected rows, and duration, so that a sudden drop in volume raises an alert before a business user finds an empty report.