Chapter 5: Streaming Ingestion and Real-Time Pipelines

Streaming Foundations and AWS Services

From Queues to Pub/Sub Logs

A stream is an unbounded sequence of events ordered (loosely) by time. A traditional queue like SQS is a one-shot mailbox — one consumer reads, the message disappears. A pub/sub log like Kafka or Kinesis is more like a bookshelf: producers append, multiple consumer groups read independently from any offset, and reading does not erase entries. Logs separate storage from position, enabling replay, multiple independent consumers, and time-travel debugging.

Producers, Partitions, and Offsets

A partition key decides which partition (Kafka/MSK) or shard (Kinesis) a record lands in. Records with the same key preserve order; different keys parallelize across partitions. Each record gets a monotonically increasing offset (Kafka) or sequence number (Kinesis) that consumers checkpoint to durable storage so they can resume after a crash.

The central trade-off: ordering is per-partition, never global, because global ordering would require a single bottleneck and destroy horizontal scalability.

Backpressure and Ordering Guarantees

When consumers fall behind producers, real systems apply backpressure: Kafka brokers raise ack latency, Kinesis throws ProvisionedThroughputExceededException, and Flink propagates fullness back through the operator DAG to the source. The "at-least-once" default means a consumer crash after processing but before checkpointing replays the record — double-counting events. Most of this chapter is about how to fix that.

The Three AWS Streaming Services

AWS exposes three first-party streaming primitives. Picking incorrectly is the most common — and most expensive — mistake in AWS streaming architecture.

Amazon Kinesis Data Streams (KDS)

A shard-based replayable log. Each shard supports 1 MB/s ingest, 1,000 records/s, and 2 MB/s egress. Default 24-hour retention, extendable to 365 days. End-to-end latency ~200 ms. The dominant failure mode is the hot shard — skewed partition keys overwhelm one shard while the rest sit idle. Native delivery is at-least-once; exactly-once requires storing sequenceNumber alongside the business write.

Amazon MSK (managed Apache Kafka)

Real Apache Kafka brokers. Full Kafka API: idempotent producers, transactions, log compaction, Kafka Streams, Kafka Connect, read_committed isolation. Native exactly-once with enable.idempotence=true + transactional writes. Latency 10–100 ms. Two flavors: Provisioned (predictable cost-efficient) and Serverless (auto-scaling, simpler billing).

Amazon Data Firehose

Not a stream — a delivery service. No consumer API, no replay. Receives records via PUT or from a Kinesis/MSK source, optionally transforms with Lambda or converts JSON→Parquet, then pushes to S3, Redshift, OpenSearch, Splunk, or HTTP endpoints. Buffer-and-flush model; latency 1–60 s. Cheapest option (~$0.029/GB). At-least-once delivery; exactly-once happens at the destination.

Diagram: AWS Streaming Fan-Out

Stream Processing with Flink

Amazon Managed Service for Apache Flink (MSF)

Apache Flink is the de-facto standard for stateful, exactly-once stream processing. Amazon Managed Service for Apache Flink (MSF) — formerly Kinesis Data Analytics for Apache Flink — runs Flink TaskManagers on AWS-managed compute units called Kinesis Processing Units (KPUs): ~1 vCPU, 4 GB RAM, 50 GB local storage each. AWS handles patching, JobManager HA, and durable checkpoint storage to S3.

Native connectors exist for KDS, MSK, Firehose, DynamoDB Streams, S3, OpenSearch, and Lambda. The default state backend is RocksDB with incremental checkpoints; the default checkpoint interval is 60 seconds.

Key CloudWatch Metrics

DataStream API vs Table / SQL API

Flink offers two programming models that compile to the same execution graph.

The DataStream API is imperative (Java/Scala/Python): you write operators (map, filter, keyBy, window, process) explicitly. Use it for complex event processing, custom state machines, side outputs, and operator-level control.

DataStream<TempReading> readings = env
    .fromSource(kinesisSource, watermarkStrategy, "iot-temperature")
    .map(json -> mapper.readValue(json, TempReading.class));

DataStream<Alert> alerts = readings
    .keyBy(TempReading::getDeviceId)
    .window(TumblingEventTimeWindows.of(Time.minutes(1)))
    .aggregate(new AvgTemperature())
    .filter(avg -> avg.value > 80.0)
    .map(avg -> new Alert(avg.deviceId, "Overheating", avg.value));

alerts.sinkTo(kinesisAlertSink);

The Table API / Flink SQL is declarative: describe the computation in SQL, and Flink's planner produces an optimal execution graph. Shorter, more maintainable, accessible to analysts.

CREATE TABLE iot_temperature (
    device_id STRING,
    ts TIMESTAMP_LTZ(3),
    temperature_c DOUBLE,
    WATERMARK FOR ts AS ts - INTERVAL '5' SECONDS
) WITH ('connector' = 'kinesis', ...);

SELECT device_id, window_start, AVG(temperature_c) AS avg_c
FROM TABLE(TUMBLE(TABLE iot_temperature, DESCRIPTOR(ts), INTERVAL '1' MINUTES))
GROUP BY device_id, window_start
HAVING AVG(temperature_c) > 80.0;

The two APIs interoperate: convert DataStream to Table and back. A common pattern: do raw deserialization and enrichment in DataStream API, expose as a Table, let analysts build downstream analytics in SQL.

State Backends and Checkpointing

Stateful streaming means operators remember things between events: a windowed aggregator stores running totals, a join stores unmatched left-side records, a session detector stores last-seen timestamps.

The Barrier-Based Checkpoint Algorithm

Flink's checkpoint is a barrier-based asynchronous distributed snapshot (a Chandy-Lamport variant):

1. The JobManager triggers a checkpoint at a configured interval (e.g., every 60 s).
2. Sources receive a numbered barrier and record their offsets/sequence numbers.
3. Barriers flow through the operator DAG alongside data, preserving order.
4. Each operator aligns its inputs (in EXACTLY_ONCE mode), then asynchronously snapshots state to the backend.
5. Operators acknowledge the checkpoint to the JobManager.
6. Once all tasks ack, the checkpoint is globally durable; sinks commit two-phase transactions.

Imagine a parade: the JobManager periodically inserts a flag-bearer (the barrier). Each viewing station (operator) waits until flag-bearers from every parallel route arrive, then photographs itself (snapshot). The photographs collectively form a consistent global snapshot.

Incremental checkpoints (RocksDB) persist only the SST files that changed since the previous snapshot, dramatically reducing checkpoint duration and S3 cost. Savepoints are user-triggered checkpoints used for upgrades, parallelism changes, and version migrations — you stop the job with a savepoint, deploy a new JAR, and restart from the savepoint with zero data loss.

Diagram: Flink Checkpoint Sequence

Time, Windows, and Correctness

Event Time vs Processing Time

Event time is the timestamp embedded in the record — when the sensor sampled, when the user clicked, when the trade was executed. Processing time is the wall-clock time when the engine sees the record. They diverge under network delay, mobile offline buffering, queue lag, or backfill.

Consider a phone buffering GPS pings while the user is in the subway. After 20 minutes underground, all pings flush at once. Processing-time view: "1,200 events arrived at 10:23 AM." Event-time view: "the user moved through these stations between 10:00 and 10:20 AM, in this order." The event-time view is correct for any meaningful analytics.

Rule of thumb: use event time for any business logic that must produce the same result if you replay yesterday's data. Use processing time only for "is the pipeline currently alive" monitoring.

Window Types: Tumbling, Sliding, Session, Global

-- Tumbling 1-hour by region
SELECT window_start, region, SUM(amount) AS revenue
FROM TABLE(TUMBLE(TABLE orders, DESCRIPTOR(order_ts), INTERVAL '1' HOUR))
GROUP BY window_start, region;

-- Sliding 5-min/1-min CPU average
SELECT window_start, AVG(cpu_pct)
FROM TABLE(HOP(TABLE host_metrics, DESCRIPTOR(ts), INTERVAL '1' MINUTE, INTERVAL '5' MINUTE))
GROUP BY window_start;

-- Session 30-min gap
SELECT user_id, window_start, window_end, COUNT(*) AS pageviews
FROM TABLE(SESSION(TABLE clickstream, DESCRIPTOR(ts), INTERVAL '30' MINUTES))
GROUP BY user_id, window_start, window_end;

Watermarks: The Engine's Promise

A watermark W(t) is a monotonically increasing assertion that "no more events with event-time timestamp ≤ t will arrive." It is how the engine knows it is safe to fire an event-time window: once W(t) advances past the window's end, no more in-time data should be coming.

Watermark propagation through multi-input operators: the operator's output watermark equals the minimum of its input watermarks. This guarantees correctness but creates the watermark stall — one slow or idle partition holds back the entire pipeline. Fix with withIdleness(...) and instrument currentInputWatermark.

Late Events: Three Treatments

1. Dropped (default in many APIs).
2. Allowed lateness via allowedLateness(Duration): window state stays alive past the watermark; each late event re-fires the window. Downstream sinks must handle multiple emissions (use upsert sinks).
3. Side output via sideOutputLateData(OutputTag): late events route to a separate stream for inspection or repair.

Diagram: Watermark-Driven Window Firing

End-to-End Exactly-Once: Three Composed Layers

1. Replayable sources — Kafka offsets and Kinesis sequence numbers stored inside the checkpoint. On recovery, the source resumes from the last committed offset.
2. Internal state — checkpoint barriers ensure operator state is consistent across the snapshot boundary.
3. Transactional or idempotent sinks — either two-phase commit (TwoPhaseCommitSinkFunction) or idempotent operations (key-based upsert).

The two-phase commit pattern:

beginTransaction() ──> invoke(record) ──> preCommit() ──> commit()
   (start of CP)        (during CP)        (CP barrier)    (CP complete)

When the barrier arrives, preCommit() flushes buffers and prepares the transaction. After notifyCheckpointComplete(checkpointId), commit() atomically exposes writes (Kafka transaction commit, S3 staging-file rename to final). On crash between pre-commit and commit, the transaction is recovered from checkpoint state and either committed or aborted on restart.

Common Flink Failure Modes