Imported from personamanagmentlayer/pcl (stdlib/data/stream-processing-expert/SKILL.md). Install upstream with npx skills add personamanagmentlayer/pcl --skill stream-processing-expert. Copyright stays with the author.

Stream Processing Expert

Continuous computation over unbounded data. The hard parts are not throughput — they are time, state and what happens when something is retried. Complements kafka-expert (the transport) and airflow-expert (batch orchestration).

Core Concepts

Event Time, Not Processing Time

Every event carries the time it happened. The stream processor sees it later, out of order, sometimes much later. Windowing on arrival time produces results that change if you replay the same data — which makes them untestable and unreproducible.

If the source has no reliable event timestamp, that is a defect in the producer. Fix it there.

Watermarks

A watermark is the processor's assertion that no event older than time T will arrive. It is how an unbounded stream produces a finite answer: when the watermark passes the end of a window, the window closes and emits.

The watermark is a bet on lateness. Too tight and correct data is dropped; too loose and every result waits.

# Flink: allow 30 seconds of out-of-orderness
watermark_strategy = (
    WatermarkStrategy
    .for_bounded_out_of_orderness(Duration.of_seconds(30))
    .with_timestamp_assigner(lambda event, _: event["occurred_at_ms"])
    .with_idleness(Duration.of_minutes(1))     # an idle partition must not stall time
)

with_idleness matters in production: a partition with no traffic holds the watermark back for the whole job, and every window stops emitting. This is a frequent and confusing outage.

Windows

Sliding windows multiply state: a 1-hour window sliding every minute keeps each event in 60 windows. Check the cost before choosing the slide.

Delivery Semantics

• At most once — losses on failure. Rarely acceptable.
• At least once — duplicates on retry. Acceptable when the sink is idempotent.
• Exactly once — no loss, no duplicates, within the system's boundaries. Achieved by checkpointed state plus a transactional or idempotent sink. It is not magic: a sink that cannot participate in a transaction cannot give it to you.

The pragmatic default: at-least-once processing with an idempotent sink keyed on a business identifier. It is simpler, cheaper and survives more failure modes than end-to-end transactions.

Flink

from pyflink.table import EnvironmentSettings, TableEnvironment

t_env = TableEnvironment.create(EnvironmentSettings.in_streaming_mode())

t_env.execute_sql("""
CREATE TABLE orders (
    order_id     STRING,
    customer_id  STRING,
    amount_minor BIGINT,
    occurred_at  TIMESTAMP_LTZ(3),
    WATERMARK FOR occurred_at AS occurred_at - INTERVAL '30' SECOND
) WITH (
    'connector' = 'kafka',
    'topic' = 'orders',
    'properties.bootstrap.servers' = 'kafka:9092',
    'properties.group.id' = 'revenue-agg',
    'scan.startup.mode' = 'group-offsets',
    'format' = 'json',
    'json.ignore-parse-errors' = 'true'
)""")

t_env.execute_sql("""
INSERT INTO revenue_by_hour
SELECT window_start, window_end, customer_id,
       SUM(amount_minor) AS revenue_minor,
       COUNT(*)          AS order_count
FROM TABLE(TUMBLE(TABLE orders, DESCRIPTOR(occurred_at), INTERVAL '1' HOUR))
GROUP BY window_start, window_end, customer_id
""")

Checkpointing is what makes state recoverable:

env.enable_checkpointing(60_000)                       # every 60s
config = env.get_checkpoint_config()
config.set_checkpointing_mode(CheckpointingMode.EXACTLY_ONCE)
config.set_min_pause_between_checkpoints(30_000)
config.set_checkpoint_timeout(600_000)
config.enable_externalized_checkpoints(
    ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION)

Retaining checkpoints on cancellation is essential — without it, stopping a job discards the state you need to resume.

Spark Structured Streaming

Micro-batch rather than true streaming; simpler to operate if you already run Spark, at the cost of latency measured in seconds.

events = (spark.readStream
    .format("kafka")
    .option("kafka.bootstrap.servers", "kafka:9092")
    .option("subscribe", "orders")
    .option("startingOffsets", "earliest")
    .option("maxOffsetsPerTrigger", 500_000)          # bound each batch
    .load()
    .select(from_json(col("value").cast("string"), SCHEMA).alias("e"))
    .select("e.*")
    .withWatermark("occurred_at", "30 seconds"))

agg = (events
    .groupBy(window("occurred_at", "1 hour"), "customer_id")
    .agg(sum("amount_minor").alias("revenue_minor")))

query = (agg.writeStream
    .outputMode("update")
    .format("delta")
    .option("checkpointLocation", "s3://lake/_checkpoints/revenue")
    .trigger(processingTime="1 minute")
    .start())

The checkpoint location is the job's identity. Changing it restarts from scratch; deleting it loses exactly-once guarantees. Never point two jobs at one checkpoint.

Output modes: append (only finalised rows, needs a watermark), update (rows that changed), complete (the whole result table — only for bounded aggregates).

Change Data Capture

CDC turns a database's write-ahead log into a stream, without polling and without touching application code.

{
  "name": "orders-connector",
  "config": {
    "connector.class": "io.debezium.connector.postgresql.PostgresConnector",
    "database.hostname": "postgres.internal",
    "database.dbname": "shop",
    "plugin.name": "pgoutput",
    "slot.name": "debezium_orders",
    "publication.autocreate.mode": "filtered",
    "table.include.list": "public.orders,public.order_lines",
    "topic.prefix": "shop",
    "snapshot.mode": "initial",
    "tombstones.on.delete": "true",
    "decimal.handling.mode": "string",
    "time.precision.mode": "connect"
  }
}

Operational facts that catch teams out:

• A replication slot retains WAL until consumed. A stopped connector fills the primary's disk. Monitor slot lag and alert well before the disk does.
• decimal.handling.mode defaults to base64-encoded bytes. Set it to string unless you enjoy debugging silently wrong money.
• The initial snapshot can be long and heavy. Plan it; consider snapshot.mode=never plus a separate bulk load for very large tables.
• Schema changes propagate. Consumers must tolerate new columns; use a schema registry with compatibility rules.
• Deletes emit a tombstone — a null value with a key. Consumers that ignore nulls silently miss deletions.

State Management

State is what makes streaming hard to operate. It grows, it must be checkpointed, and it must be restorable.

• Bound it. Every keyed state needs a TTL, or the job grows until it dies.
• Choose the backend by size. Heap state is fast and bounded by memory; RocksDB spills to disk and supports far larger state with more latency.
• Take savepoints before every deploy. Savepoints are the mechanism for upgrading a stateful job without losing its accumulated state.
• Version your state schema. Adding a field to a stored object breaks restoration unless the serialiser supports evolution.

state_descriptor.enable_time_to_live(
    StateTtlConfig.new_builder(Time.days(7))
        .set_update_type(StateTtlConfig.UpdateType.OnCreateAndWrite)
        .cleanup_in_rocksdb_compact_filter(1000)
        .build()
)

Late and Out-of-Order Data

-- Flink: keep windows open past the watermark for stragglers
SELECT window_start, SUM(amount_minor)
FROM TABLE(TUMBLE(TABLE orders, DESCRIPTOR(occurred_at), INTERVAL '1' HOUR))
GROUP BY window_start
-- with table.exec.emit.allow-lateness = 1h in configuration

Route data later than your allowed lateness to a side output rather than dropping it silently. Counting what you discard is how you discover the watermark is wrong.

The general answer for correctness over long horizons is a batch reconciliation job: the stream gives low latency, a nightly batch recomputes the same aggregate from the source of truth, and any divergence is an alert. Streaming alone should not be the system of record for financial figures.

Best Practices

• Key by a business identifier so partitioning is stable and rebalances do not reorder related events.
• Make sinks idempotent — upsert on a natural key rather than insert.
• Bound every batch (maxOffsetsPerTrigger, scan.parallelism) so a backlog does not produce one enormous batch that fails repeatedly.
• Monitor lag, not throughput. Consumer lag and watermark delay are the signals that matter.
• Replay-test. Reprocess a day of historical data and compare against batch output; if they disagree, the pipeline is not deterministic.
• Handle poison messages with a dead-letter topic and a parse-error policy; one malformed event must not stop the job forever.
• Deploy with savepoints, always.

Anti-Patterns

• Windowing on processing time for business aggregates — results change on replay.
• Unbounded state with no TTL — a slow, certain outage.
• Ignoring an idle partition — the watermark stalls and nothing emits.
• Assuming exactly-once from configuration alone — the sink must participate.
• CDC without monitoring replication slot lag — fills the primary's disk.
• Sharing a checkpoint directory between jobs.
• Streaming as the ledger with no batch reconciliation.
• Reaching for streaming when a five-minute batch would do — the operational cost is an order of magnitude higher.

Reference Documentation

• Operations — deployment, savepoint upgrades, backpressure diagnosis, scaling, monitoring and failure playbooks

Resources

• Apache Flink documentation
• Spark Structured Streaming guide
• Debezium documentation
• Tyler Akidau et al., Streaming Systems
• Martin Kleppmann, Designing Data-Intensive Applications, chapter 11