Real-Time Stream Processing with PyFlink

A stream-processing pipeline built on Apache Flink 1.20 (PyFlink DataStream API) that ingests three live streams, joins and enriches them, tracks rolling-window metrics, detects anomalies, handles late events without dropping them, and raises live pipeline health alerts. It demonstrates the core primitives that production stream processing depends on: event-time semantics with watermarks, keyed partitioning, a stateful two-stream join with timers, sliding and tumbling windowed aggregation, keyed-state anomaly detection, side outputs for late data, and a self-monitoring health layer.

The pipeline ingests three sources:

• ORDERS: order_id, customer_id, order_amount, order_ts, country
• PAYMENTS: payment_id, order_id, paid_amount, payment_ts, method
• DEVICE EVENTS: device_id, site_id, event_type, status, event_ts, where event_type is one of heartbeat, fault, reboot, or firmware_update

It runs five families of processing on top of those sources.

Stage 1, join and enrichment. Orders and payments are keyed by order_id and joined within an event-time window by a stateful keyed co-process function that buffers whichever side arrives first and arms an event-time timer. The output is a stream of enriched payment records carrying the payment fields plus the matching order context, with two flags: amount_mismatch when the paid amount differs from the order amount, and unmatched when a payment never matched an order within the window.

Stage 2, rolling-window metrics. Sliding and tumbling event-time windows summarize the streams as they flow. Enriched payments roll up per tumbling window into joined count, unmatched payments, amount mismatches, and total revenue. Device events roll up per site over a sliding window into event count, fault count, reboot count, firmware failures, and the fault rate.

Stage 3, anomaly detection. Enriched, matched payments are keyed by customer and scored for two anomaly classes:

• Amount anomaly: a payment whose value sits far above the customer's own typical spend. Each customer maintains a running mean and standard deviation of its payment amounts in keyed state, and a payment is flagged when it exceeds a dynamic threshold of mean plus three standard deviations. A static floor guards the cold-start period before enough history has accumulated.
• Velocity anomaly: more payments than an allowed maximum inside a short sliding window for a single customer, the signature of card-testing or abuse.

Device events are keyed by site or device and scored for three more classes:

• Device fault spike: a site whose fault count inside a sliding window crosses a threshold, the signature of a site-wide outage.
• Reboot loop: a device that power-cycles more than an allowed number of times inside a window, the signature of a crash loop.
• Firmware-update failure: a device that reports failed firmware updates, surfaced immediately from keyed state with a running failure count.

Stage 4, late-event handling. Every stream carries event-time watermarks with a bounded out-of-orderness. The device stream adds an explicit allowed lateness on top of that bound. An event that falls behind the effective watermark by more than the allowed lateness is not dropped silently: a keyed late-event router sends it to a side output identified by an OutputTag, tagged with the watermark that judged it late and the resulting lag, so late events are counted and inspected rather than lost.

Stage 5, live pipeline health alerts. A health layer watches the pipeline itself. It fuses per-event throughput signals with the late-event side output and raises an alert whenever health degrades: a throughput stall between consecutive events, a late-event spike, a watermark lag beyond a threshold, or a fault rate over a ceiling. Alerts are emitted to a print sink and, optionally, a rolling file sink standing in for an alerting channel.

Architecture

flowchart LR
    O[Order source] --> W[Event-time<br/>watermarks]
    P[Payment source] --> W
    D[Device source] --> W

    W --> KO[keyBy order_id]
    KO --> J[Keyed co-process join<br/>state plus event-time timers]
    J --> E[Enriched payments<br/>order context, mismatch, unmatched]

    E --> M[Tumbling window metrics<br/>joined, unmatched, revenue]
    E --> KC[keyBy customer_id]
    KC --> PA[Payment anomalies<br/>amount rule, velocity rule]

    W --> LR[Late-event router<br/>allowed lateness]
    LR -->|on time| DM[Sliding window metrics<br/>per-site fault rate]
    LR -->|on time| DA[Device anomalies<br/>fault spike, reboot loop, firmware fail]
    LR -->|late| LO[Late-event side output<br/>OutputTag, counted]

    LR --> H[Health monitor]
    LO --> H
    H --> HA[Health alerts<br/>throughput, late rate, watermark lag, fault rate]

    E --> S[Sinks<br/>print and optional file]
    M --> S
    PA --> S
    DM --> S
    DA --> S
    LO --> S
    HA --> S

Loading

Orders and payments are keyed by order_id so the join co-locates the two sides of each transaction; the enriched stream then fans out to tumbling-window metrics and to per-customer anomaly detection. Device events pass through a late-event router that diverts late data to a side output and forwards on-time data to per-site rolling metrics and per-site or per-device anomaly detection. The health monitor consumes throughput signals and the late-event side output to alert on pipeline degradation.

Tech stack

• Apache Flink 1.20.0 via PyFlink (DataStream API)
• Python 3.11
• Java 17 (Java 11 or 17 recommended for the Flink runtime; see the note below)
• Docker and Docker Compose for the standalone JobManager and TaskManager cluster

Project layout

real-time-stream-processing/
  src/job.py                 PyFlink job: three sources, join, metrics, anomalies, late handling, health
  src/enrichment.py          Keyed co-process join, enrichment, payment window metrics
  src/devices.py             Device metrics, device anomaly detectors, late-event router
  src/health.py              Windowed health monitor and gap-based throughput monitor
  scripts/generate_stream.py Deterministic order, payment, and device-event generators
  Dockerfile                 Flink 1.20 image extended with Python and PyFlink
  docker-compose.yml         JobManager plus TaskManager cluster
  requirements.txt           Pinned to apache-flink==1.20.0

Tuning

The join parameters live at the top of src/enrichment.py:

Parameter	Default	Meaning
JOIN_WINDOW_SECONDS	10	How long to wait for the matching event, event time
AMOUNT_MATCH_TOLERANCE	0.01	Allowed paid-vs-order difference before mismatch

The payment metrics and anomaly parameters live at the top of src/job.py:

Parameter	Default	Meaning
METRICS_WINDOW_SECONDS	10	Tumbling window length for running payment metrics
WINDOW_SIZE_SECONDS	5	Sliding window length for the velocity rule
WINDOW_SLIDE_SECONDS	1	How often the velocity window advances
MAX_TRANSACTIONS_PER_WINDOW	4	Velocity threshold per customer per window
AMOUNT_SIGMA_MULTIPLIER	3.0	Standard deviations above the running mean
AMOUNT_STATIC_FLOOR	5000.0	Cold-start amount threshold
MIN_SAMPLES_FOR_DYNAMIC	5	Samples required before the dynamic threshold applies
ALLOWED_LATENESS_SECONDS	2	Out-of-order tolerance for watermarks

The device parameters live at the top of src/devices.py:

Parameter	Default	Meaning
DEVICE_WINDOW_SIZE_SECONDS	5	Sliding window length for device metrics and rules
DEVICE_WINDOW_SLIDE_SECONDS	1	How often the device window advances
MAX_FAULTS_PER_WINDOW	3	Fault-spike threshold per site per window
MAX_REBOOTS_PER_WINDOW	3	Reboot-loop threshold per device per window
OUT_OF_ORDER_BOUND_MS	2000	Out-of-order bound mirrored by the late router
DEVICE_ALLOWED_LATENESS_MS	2000	Extra lateness before an event is diverted late

The health parameters live at the top of src/health.py:

Parameter	Default	Meaning
HEALTH_WINDOW_SECONDS	5	Tumbling window length for health evaluation
MAX_LATE_EVENTS_PER_WINDOW	0	Late-event count ceiling before an alert
MAX_WATERMARK_LAG_MS	5000	Watermark-lag ceiling before a critical alert
MAX_FAULTS_PER_WINDOW	4	Fault count ceiling before a critical alert
MAX_EVENT_GAP_MS	4000	Inter-event gap before a throughput alert

Run option 1: local mini-cluster

This is the fastest way to see the pipeline work. PyFlink spins up an embedded mini-cluster inside the Python process. The job uses bounded, deterministic order, payment, and device datasets that always contain a matched pair for the join, one planted amount mismatch, one planted unmatched payment, one planted large-amount anomaly, one planted velocity burst, one planted device fault spike, one planted reboot loop, one planted firmware-failure run, one planted late event, and one planted throughput drop, so every stage fires on every run.

cd real-time-stream-processing
/opt/homebrew/bin/python3.11 -m venv .venv
./.venv/bin/pip install --upgrade "setuptools<66" wheel
./.venv/bin/pip install -r requirements.txt
PYTHONPATH=. ./.venv/bin/python src/job.py

To also persist metrics, anomalies, and health alerts to disk, set an output directory:

ANOMALY_OUTPUT_DIR=./output PYTHONPATH=. ./.venv/bin/python src/job.py

If Flink resolves only python3 on your PATH, the job already points the Python worker at the current interpreter. Override it with PYFLINK_PYTHON if needed:

PYFLINK_PYTHON=$(pwd)/.venv/bin/python PYTHONPATH=. ./.venv/bin/python src/job.py

Run option 2: standalone Flink cluster with Docker

The compose file builds a Flink 1.20 image with Python and PyFlink installed, then starts a JobManager and a TaskManager. The project directory is mounted into the containers so the job can be submitted directly.

cd real-time-stream-processing
docker compose up -d --build

The Flink web dashboard is available at http://localhost:8081. Submit the job to the running cluster:

docker compose exec jobmanager \
  flink run -py /opt/flink/usrlib/src/job.py \
  -pyfs /opt/flink/usrlib

The -pyfs /opt/flink/usrlib argument puts the mounted project root on the Python path so the src and scripts packages resolve inside the cluster.

Output prints to the TaskManager task output, viewable in the dashboard or via logs:

docker compose logs -f taskmanager

Tear the cluster down when finished:

docker compose down

Streaming continuous sources

scripts/generate_stream.py can emit newline-delimited JSON for any of the three streams to standard output at a configurable rate, which is useful for wiring the job to socket or file sources or for feeding a message broker:

./.venv/bin/python scripts/generate_stream.py --stream orders --rate 20 --seed 7
./.venv/bin/python scripts/generate_stream.py --stream payments --rate 20 --seed 7
./.venv/bin/python scripts/generate_stream.py --stream devices --rate 20 --seed 7

An order, a payment, and a device record have these shapes:

{"order_id": "order-0000", "customer_id": "cust-003", "order_amount": 143.22, "order_ts": 1700000000000, "country": "CA"}
{"payment_id": "pay-0000", "order_id": "order-0000", "paid_amount": 143.22, "payment_ts": 1700000000866, "method": "card"}
{"device_id": "dev-100", "site_id": "site-a", "event_type": "fault", "status": "degraded", "event_ts": 1700000030000}

Sample output

Running the local mini-cluster prints tagged, newline-delimited JSON for each stage. The lines below are copied from a real run.

Enriched payments from the join, including the planted mismatch and the unmatched payment:

ENRICHED> {"record_type": "enriched_payment", "payment_id": "pay-9001", "order_id": "order-9001", "paid_amount": 90.0, "method": "card", "payment_ts_ms": 1700000026157, "unmatched": false, "customer_id": "cust-003", "order_amount": 240.0, "country": "US", "amount_mismatch": true}
ENRICHED> {"record_type": "enriched_payment", "payment_id": "pay-9002", "order_id": "order-DOES-NOT-EXIST", "paid_amount": 55.0, "method": "wallet", "payment_ts_ms": 1700000026557, "unmatched": true, "customer_id": null, "order_amount": null, "country": null, "amount_mismatch": false}

Rolling-window metrics. A tumbling window over enriched payments, and a sliding window over device events per site catching the fault spike as it builds:

METRICS> {"record_type": "window_metrics", "window_start_ms": 1700000010000, "window_end_ms": 1700000020000, "joined_count": 12, "unmatched_payments": 0, "amount_mismatches": 0, "total_revenue": 965.6}
DEVICE_METRICS> {"record_type": "device_metrics", "site_id": "site-b", "window_start": "2023-11-14T22:13:51+00:00", "window_end": "2023-11-14T22:13:56+00:00", "event_count": 10, "fault_count": 8, "reboot_count": 0, "firmware_failures": 0, "fault_rate": 0.8}

Anomalies. The planted large-amount payment on cust-001 and cust-002's velocity burst, then the three device anomaly classes:

ANOMALY> {"record_type": "anomaly", "anomaly_type": "amount", "customer_id": "cust-001", "payment_id": "pay-9003", "order_id": "order-9003", "amount": 9800.0, "threshold": 130.87, "event_time": "2023-11-14T22:13:48.057000+00:00"}
ANOMALY> {"record_type": "anomaly", "anomaly_type": "velocity", "customer_id": "cust-004", "transaction_count": 5, "total_amount": 430.52, "window_start": "2023-11-14T22:13:33+00:00", "window_end": "2023-11-14T22:13:38+00:00"}
DEVICE_ANOMALY> {"record_type": "anomaly", "anomaly_type": "device_fault_spike", "site_id": "site-b", "fault_count": 6, "threshold": 3, "window_start": "2023-11-14T22:13:50+00:00", "window_end": "2023-11-14T22:13:55+00:00"}
DEVICE_ANOMALY> {"record_type": "anomaly", "anomaly_type": "device_reboot_loop", "device_id": "dev-300", "reboot_count": 4, "threshold": 3, "window_start": "2023-11-14T22:13:52+00:00", "window_end": "2023-11-14T22:13:57+00:00"}
DEVICE_ANOMALY> {"record_type": "anomaly", "anomaly_type": "firmware_update_failure", "device_id": "dev-100", "site_id": "site-a", "failure_count": 1, "event_time": "2023-11-14T22:13:57.507000+00:00"}

Late-event handling. The planted late device event, captured by the side output instead of being dropped, tagged with the watermark that judged it late and the resulting lag:

LATE_EVENT> {"record_type": "device_event", "device_id": "dev-101", "site_id": "site-a", "event_type": "heartbeat", "status": "ok", "event_ts_ms": 1700000003507, "watermark_ms": 1700000039650, "lag_ms": 36143}

Live pipeline health alerts. All four planted degradation scenarios fire: the throughput gap, the late-event spike, the watermark lag, and the fault rate:

HEALTH_ALERT> {"record_type": "health_alert", "reason": "throughput_below_floor", "severity": "warning", "gap_ms": 12650, "threshold_ms": 4000, "resumed_at": "2023-11-14T22:13:51.607000+00:00"}
HEALTH_ALERT> {"record_type": "health_alert", "reason": "late_event_spike", "severity": "warning", "window_start": "2023-11-14T22:13:20+00:00", "window_end": "2023-11-14T22:13:25+00:00", "late_events": 1, "ceiling": 0}
HEALTH_ALERT> {"record_type": "health_alert", "reason": "watermark_lag_exceeded", "severity": "critical", "window_start": "2023-11-14T22:13:20+00:00", "window_end": "2023-11-14T22:13:25+00:00", "max_lag_ms": 36143, "threshold_ms": 5000}
HEALTH_ALERT> {"record_type": "health_alert", "reason": "fault_rate_exceeded", "severity": "critical", "window_start": "2023-11-14T22:13:50+00:00", "window_end": "2023-11-14T22:13:55+00:00", "fault_count": 6, "ceiling": 4}

Reading the output top to bottom: the join flags a payment settled for 90.0 against a 240.0 order and a payment whose order never arrived; the tumbling and sliding windows roll live traffic into per-window counts, revenue, and per-site fault rate; the anomaly detectors catch the large payment, the velocity burst, the site fault spike, the device reboot loop, and the firmware failures; the late router captures a device event that arrived 36 seconds behind the watermark; and the health monitor pages on the throughput stall, the late event, the watermark lag, and the fault rate.

Note on Java version

PyFlink 1.20 officially targets Java 11 and Java 17. The job was developed and run successfully on the embedded mini-cluster under Java 21, and the Docker cluster image pins Java 17 for the runtime. If a local run fails to start the mini-cluster on a newer JDK, install a Java 17 runtime and point JAVA_HOME at it before running. The Python sources are otherwise interpreter-agnostic and byte-compile cleanly.

All glory to God! ✝️❤️