Chapter 2 — Data Ingestion: Sources, Formats, and Patterns

ML pipelines rarely consume data from a single tidy source. A production fraud-detection model might pull transaction events from Kafka, account state from a Postgres database via change data capture, merchant metadata from a nightly S3 export, and risk-list updates from a third-party REST API. Each source has a native shape, a natural cadence, and a preferred ingestion mechanism. Forcing one paradigm onto all sources is the most common ingestion anti-pattern.

Relational databases and CDC. The simplest path — periodic SELECTs scheduled by Airflow — scales poorly: nightly scans waste compute when few rows changed, and they load a database that is also serving production traffic. CDC solves this by reading the transaction log directly. Debezium tails the MySQL binlog or the Postgres WAL, converts inserts/updates/deletes into structured events, and publishes them to Kafka. CDC preserves source-of-truth semantics while enabling incremental updates; the cost is operational complexity (log permissions, schema evolution, periodic full re-syncs).

Object storage, data lakes, and lakehouses. S3, GCS, and ADLS host files from batch ETL jobs, third-party providers, and log shippers. A data lake is permissive: any team can drop any file. A lakehouse layers a table format (Delta Lake, Iceberg, Hudi) on top of Parquet, giving the collection transactional semantics, schema evolution, and time travel. ML teams commonly use bronze (raw), silver (cleaned), and gold (feature-ready) zones, training on gold and snapshotting at a known version for reproducibility.

Event streams. Apache Kafka is the de facto self-managed standard; AWS Kinesis, Google Pub/Sub, and Azure Event Hubs are managed equivalents. Streams matter to ML for two reasons: they are the substrate of real-time features (e.g., "items viewed in the last 5 minutes"), and they serve as a durable buffer between source systems and downstream processors — a fraud pipeline can fall behind for an hour during a deploy without losing data.

APIs, logs, and sensors. External HTTP APIs (weather, FX, risk scores) need rate-limit awareness and retry logic. Application logs (shipped via Fluentd or Logstash) need parsing and timestamp normalization. IoT sensor feeds (often MQTT before crossing into Kafka) need gap detection and clock-skew handling.

Figure 2.1: Heterogeneous data sources feeding an ML ingestion layer

Once you know where data lives, the next question is how often to move it. The decision is driven by one variable above all: how stale can features be before model quality suffers?

Batch patterns. An orchestrator (Airflow, Prefect, Dagster) triggers a Spark or SQL job on a cron schedule. The job reads a delta, transforms it, and writes to a destination table. Batch is ideal for (a) building large historical training sets from a lake, (b) computing slow-moving aggregates (30-day spend, lifetime value), and (c) backfills when a new feature definition must be applied to historical data.

Streaming with Kafka and Flink. A producer publishes to Kafka or Kinesis; a stream processor (Flink, Spark Structured Streaming, Kafka Streams) consumes events, applies windowed transformations, and writes results to an online store. Online stores (Redis, DynamoDB, Cassandra, Aerospike) serve features at single-digit-ms latency for inference. Streaming is the right choice when models must react within the same session.

Figure 2.2: Batch vs streaming ingestion paths

Lambda and Kappa Architectures

Lambda maintains separate batch and speed layers. The batch layer computes accurate historical features into the offline store; the speed layer computes approximate recent features from a stream into the online store; the two are merged at serving time. Each layer uses its optimal tool but two code paths can drift in subtle ways — the canonical Lambda failure mode.

Kappa takes a different approach: a single streaming pipeline is the source of truth; both historical and real-time processing run through the same code, with history reconstructed by replaying the log. Feature logic is implemented exactly once. The cost is that replaying a year of log to backfill a new feature is expensive and operationally tricky.

Figure 2.3: Lambda architecture with batch, speed, and serving layers

In practice, vendor feature stores (Feast, Tecton, SageMaker Feature Store) implement a Lambda-like dual store but mitigate drift by letting users declare feature logic once in a DSL that generates both pipelines.

The file format you choose for training data is not a serialization detail — it determines I/O throughput, storage cost, schema-evolution flexibility, and how easily teams can share data. ML workloads have particular characteristics (wide rows, repeated reads, mixed access across frameworks) that interact with format choice in non-obvious ways.

Row-based: CSV, JSON, Avro. CSV and JSON are ubiquitous but inefficient: text wastes space, parsing is CPU-heavy, and there is no native schema enforcement. They are fine for ad-hoc inspection; they are wrong for production ML. Apache Avro is the serious row-based format: compact binary with the schema stored separately (typically in a Confluent Schema Registry), enabling field-by-field deserialization. Avro's defining strength is schema evolution — writer and reader schemas reconcile at read time, supporting added fields with defaults and renamed fields via aliases. This makes Avro the canonical format for Kafka topics, where event schemas evolve over years.

Columnar: Parquet and ORC. Columnar formats store all values of a column contiguously, which is transformative for analytical workloads. They share three key advantages: high compression (similar values pack well together), predicate pushdown (skip whole row groups based on column statistics), and column pruning (read only the columns the query needs). Parquet is the dominant choice in modern lakehouses with first-class Spark, Trino, Snowflake, and PyArrow integration. For ML, Parquet is almost always right for silver and gold layers — feature tables, training datasets, and the offline tier of a feature store.

Schema evolution on raw Parquet is workable for adds but tricky for drops or type changes. ML teams delegate evolution to a table format layered on top: Delta Lake, Iceberg, or Hudi tracks versioned schemas, supports MERGE INTO operations, and enables time travel for training reproducibility.

TFRecord and Petastorm. TFRecord is TensorFlow's native training format: a sequence of length-prefixed protobuf tf.train.Example records, optimized for sequential reads via tf.data.TFRecordDataset. When the input pipeline is the GPU-throughput bottleneck, TFRecord can be more stable than reading Parquet through a Python adapter. The cost is significant: the schema lives in parsing code (not the file), Spark integration is awkward, and cross-framework reuse is painful. Most teams should use TFRecord only as a derived training artifact materialized from Parquet for a specific TensorFlow job. Petastorm bridges Parquet and deep-learning frameworks, exposing a Parquet dataset as a streaming PyTorch/TensorFlow dataset and removing much of the motivation for TFRecord.

Compression tradeoffs. Stronger compression (Gzip, ZSTD high level) reduces storage and network cost but raises CPU on every read. Snappy and LZ4 are common defaults for ML training data because read CPU often matters more than storage. ZSTD is a strong middle ground — compression close to Gzip with decompression speed close to Snappy.

A pipeline that ingests correctly 99% of the time is not 1% wrong — it is broken. The 1% manifests as silent feature drift, training-serving skew, and missing labels that degrade accuracy in ways that are nearly impossible to debug after the fact. Reliable ingestion rests on four pillars: idempotency, schema management, backpressure handling, and lineage.

Idempotency and Exactly-Once Semantics

Distributed systems fail. Network calls time out, brokers restart, consumers get rebalanced. Every reliable pipeline must assume retries will happen and ensure that processing the same message twice produces the same result as processing it once. Within Kafka, enable.idempotence=true on a producer assigns it a producer ID and tracks sequence numbers per partition, so a retry after an in-flight failure does not duplicate the message. For read-process-write pipelines that stay inside Kafka, transactional producers go further: initTransactions(), beginTransaction(), sendOffsetsToTransaction(), commitTransaction() make output records and offset commits atomic, with consumers in read_committed mode seeing only committed transactions.

Outside Kafka, transactional guarantees do not extend — a feature store or a lake cannot participate in a Kafka transaction. The practical pattern is at-least-once delivery from Kafka combined with idempotent writes at the sink. Every event carries a stable identifier (a UUID assigned upstream, or a hash of entity ID and event time), and the sink performs upserts keyed by that identifier. Writing the same event twice overwrites the same row with the same value; the duplicate is invisible downstream.

For lake ingestion, the idiom is MERGE INTO on a Delta/Iceberg/Hudi table keyed by event ID. For online feature stores, it is SET keyed by (entity_id, feature_name) with a write-time check that ignores updates with timestamps older than the current value, preventing out-of-order events from overwriting fresh data with stale data. Compacted Kafka topics provide a third option: keyed by entity, the broker retains only the latest value per key.

Analogy: idempotency is like a hotel reservation confirmation number. If your booking app crashes and you click "Reserve" again, the hotel uses the confirmation number to recognize the duplicate and charges you once, not twice.

Figure 2.4: Kafka exactly-once flow across producer, broker, and consumer

Schema Evolution

Source schemas change — a microservice adds a field, a type widens, a nullable column becomes required. If ingestion treats every change as breaking, every minor source update halts the pipeline. The pattern is a schema registry: Confluent Schema Registry (and compatibles) stores Avro/Protobuf/JSON Schema definitions keyed by topic/subject and enforces compatibility (backward, forward, full) on every new version. Producers register the schema before publishing; consumers fetch the writer's schema by ID and reconcile with their reader's schema at deserialization.

Practical rules: add new fields with defaults (backward compatible); avoid renames (use aliases); never drop required fields; never change a field's type incompatibly. ML pipelines should always include stable identifiers — entity ID, event ID, event timestamp — as required fields, because these are the keys on which idempotency depends. On the lake side, table formats (Delta, Iceberg, Hudi) extend schema evolution to columnar files as metadata operations that do not rewrite historical data.

Backpressure, Retries, and Dead Letter Queues

A pipeline that ingests faster than its sink can absorb does not just slow down — it falls over. Memory fills, GC pauses extend, consumers get evicted, and the lag chart turns into a wall. Backpressure is the mechanism by which a slow downstream signals an upstream producer to slow down. In Kafka consumers, backpressure is largely manual: tune max.poll.records, max.partition.fetch.bytes, and fetch.max.bytes so each poll returns only what the processor can handle before the next deadline. In stream-processing frameworks (Flink, Spark Structured Streaming, Kafka Streams), backpressure is automatic.

When the sink fails on a specific record — bad schema, business-rule violation, missing reference data — the answer is a dead letter queue. The failing message is published to a side topic with error metadata; the main pipeline keeps moving. Operators triage the DLQ separately, often replaying records back to the main topic after fixing the underlying issue. For transient errors (5xx, rate limits), exponential backoff with jitter prevents thundering-herd retries that turn a brief outage into a sustained one.

Lineage

Lineage is knowing, for any training row or feature value, exactly which source events produced it, through which transformations, at which versions. It matters for three reasons: debugging (trace a suspicious feature to its source), compliance (regulated industries need to reproduce any prediction), and reproducibility (retraining last quarter's model requires reconstructing last quarter's data). Lineage is captured at multiple layers: table formats record version histories; orchestrators record job inputs/outputs; OpenLineage, Marquez, and DataHub aggregate signals into a graph; MLflow and feature stores like Feast record the dataset versions and feature definitions used in each training run.