Ingestion is how data enters your platform — pulled in batches, streamed as events, or captured from a database’s change log. Get it reliable and idempotent.
Why: ingestion is the first stage of every pipeline, and there are three main patterns — batch pull (query a source on a schedule), streaming (consume events continuously), and change data capture (read a database’s change log) — each with different freshness and complexity. When: batch for periodic loads, streaming for real-time needs, CDC to mirror a database without hammering it. Where: choosing the pattern sets the latency and cost of everything downstream.
BATCH PULL query source on a schedule simple, minutes-hours old
STREAMING consume events as they happen complex, seconds fresh
CDC read the DB's change log (WAL) mirror a DB in near-real-time
(Debezium -> Kafka)
Pick by required freshness. Most ML training ingestion is batch pull.Why: re-pulling an entire source every run wastes time and money, so batch ingestion should be incremental — fetch only rows newer than the last run (a "high-water mark") — and land them as partitioned Parquet. When: use incremental loads for any large, append-mostly source. Where: track the watermark durably so a restart resumes correctly.
import pandas as pd
from sqlalchemy import create_engine
engine = create_engine("postgresql://user:pass@db/app")
def ingest_incremental(last_ts: str, run_date: str):
# Pull ONLY new rows since the last successful run:
df = pd.read_sql(
"SELECT * FROM events WHERE updated_at > %(ts)s",
engine, params={"ts": last_ts},
)
# Land as partitioned Parquet on the lake (idempotent per run_date):
df.to_parquet(f"s3://lake/events/dt={run_date}/data.parquet")
return df["updated_at"].max() # the new high-water mark to persistWhy: bad data caught at ingestion is cheap; bad data discovered in a trained model is expensive — so validate schema, null rates, and ranges the moment data arrives and fail loudly on violations. When: run checks before writing to the lake, so corrupt data never enters the pipeline. Where: this is the same discipline as the Feature Pipeline checklist — schema and quality gates at ingestion.
def validate(df):
# Schema: required columns present
required = {"user_id", "event", "updated_at"}
missing = required - set(df.columns)
assert not missing, f"missing columns: {missing}"
# Quality: null rate and value sanity
assert df["user_id"].notna().all(), "null user_id"
assert df["updated_at"].notna().mean() > 0.99, "too many null timestamps"
return df
# Fail the run here, loudly — never let unvalidated data into the lake.