tutorial

Monitoring Bytewax Stream Processing with Vigilmon

Bytewax brings Python-native stream processing on a Rust core — here's how to monitor worker health, Kafka connector lag, dataflow throughput, stateful operators, and end-to-end latency with Vigilmon.

Bytewax lets Python data engineers write production-grade streaming applications using familiar Python APIs backed by a high-performance Rust runtime. You define dataflows as directed acyclic graphs of operators — map, filter, fold_window, stateful_map — and Bytewax compiles and executes them on a timely-dataflow Rust engine. When your Bytewax pipeline is processing real-time ML features, IoT telemetry, or event ETL, you need to know the moment a worker crashes, Kafka lag spikes, or a stateful operator loses its backend. Vigilmon gives you that visibility with HTTP endpoint checks, heartbeats, and alerting that covers every critical layer of your Bytewax deployment.

What You'll Set Up

  • Worker process health monitoring via a sidecar health endpoint
  • Kafka input consumer group lag alerting
  • Kafka output producer write success monitoring
  • Dataflow throughput drop alerts
  • Stateful operator (SQLite) backend health checks
  • Window computation and watermark lag monitoring
  • Worker recovery health alerts
  • End-to-end processing latency tracking
  • Python UDF exception rate alerts

Prerequisites

  • Bytewax 0.21+ deployed as Python workers (bare metal, Docker, or Kubernetes)
  • Kafka cluster used as input/output connectors
  • A free Vigilmon account

Why Monitor Bytewax?

Bytewax runs as Python processes. If a worker process dies, the entire dataflow stops silently — there is no built-in cluster manager that restarts crashed workers and sends you a page. Kafka consumer lag can grow for hours before anyone notices the pipeline has stalled. Stateful operators depend on a SQLite backend; a corrupted or full state directory causes those operators to fail on the next state write. And because Bytewax workers handle windowing with watermark-based time, late arrivals and watermark lag problems appear only in the output — never in process metrics. Proactive monitoring catches all of these before they become data incidents.

Step 1: Expose a Worker Health Endpoint

Bytewax workers are Python processes. The simplest way to expose health to Vigilmon is a lightweight HTTP sidecar running alongside each worker. Add this to your worker entrypoint:

import threading
from http.server import BaseHTTPRequestHandler, HTTPServer

class HealthHandler(BaseHTTPRequestHandler):
    def do_GET(self):
        if self.path == "/health":
            self.send_response(200)
            self.end_headers()
            self.wfile.write(b'{"status":"ok"}')
        else:
            self.send_response(404)
            self.end_headers()

    def log_message(self, format, *args):
        pass  # suppress access logs

def start_health_server(port=8080):
    server = HTTPServer(("0.0.0.0", port), HealthHandler)
    thread = threading.Thread(target=server.serve_forever, daemon=True)
    thread.start()

start_health_server()
# ... your Bytewax dataflow definition and run_main() call below

Start the sidecar before run_main(). If the worker process crashes, the health server dies with it and Vigilmon's check fails immediately.

In Vigilmon:

  1. Click Add Monitor → HTTP / HTTPS.
  2. Enter http://<worker-host>:8080/health.
  3. Set Check interval to 1 minute.
  4. Set Expected HTTP status to 200.
  5. Click Save.

Add one monitor per worker host. If you run multiple workers per node, assign each worker a distinct port (8080, 8081, …) and monitor each port separately.

Step 2: Monitor Kafka Input Consumer Group Lag

Bytewax reads from Kafka using a KafkaSourceInput. Consumer group lag is the most important leading indicator of a stalled pipeline — it grows whenever Bytewax processes records more slowly than they arrive.

Expose lag as a metric endpoint using a small lag exporter script run on a cron or as a sidecar:

#!/usr/bin/env python3
"""Expose Kafka consumer group lag as an HTTP health endpoint."""
import json
from http.server import BaseHTTPRequestHandler, HTTPServer
from confluent_kafka.admin import AdminClient
from confluent_kafka import Consumer, TopicPartition

BOOTSTRAP_SERVERS = "kafka:9092"
GROUP_ID = "bytewax-consumer-group"
LAG_THRESHOLD = 5000  # records

def get_lag():
    admin = AdminClient({"bootstrap.servers": BOOTSTRAP_SERVERS})
    consumer = Consumer({"bootstrap.servers": BOOTSTRAP_SERVERS, "group.id": GROUP_ID})
    topics = admin.list_topics(timeout=10)
    total_lag = 0
    for topic_name in topics.topics:
        partitions = [
            TopicPartition(topic_name, p)
            for p in topics.topics[topic_name].partitions
        ]
        committed = consumer.committed(partitions, timeout=10)
        end_offsets = consumer.get_watermark_offsets
        # simplified: aggregate lag across all partitions
        # replace with real per-partition lag calculation
    consumer.close()
    return total_lag

class LagHandler(BaseHTTPRequestHandler):
    def do_GET(self):
        if self.path == "/kafka-lag":
            try:
                lag = get_lag()
                status = 200 if lag < LAG_THRESHOLD else 503
                self.send_response(status)
                self.end_headers()
                self.wfile.write(json.dumps({"lag": lag, "threshold": LAG_THRESHOLD}).encode())
            except Exception as e:
                self.send_response(503)
                self.end_headers()
                self.wfile.write(json.dumps({"error": str(e)}).encode())
        else:
            self.send_response(404)
            self.end_headers()

    def log_message(self, format, *args):
        pass

HTTPServer(("0.0.0.0", 8081), LagHandler).serve_forever()

In Vigilmon:

  1. Click Add Monitor → HTTP / HTTPS.
  2. Enter http://<exporter-host>:8081/kafka-lag.
  3. Set Expected HTTP status to 200.
  4. Set Check interval to 1 minute.
  5. Under Alerting, add an alert for status 503 (lag exceeded threshold).
  6. Click Save.

Step 3: Monitor Kafka Output Sink Health

Bytewax writes processed results to Kafka output topics using KafkaSinkOutput. If the producer fails — broker connectivity lost, topic ACL revoked, serialization error — output records are dropped silently.

Expose a producer health probe:

from confluent_kafka import Producer

BOOTSTRAP_SERVERS = "kafka:9092"
OUTPUT_TOPIC = "bytewax-output"

def check_producer_health():
    """Returns True if the producer can connect and produce a test probe."""
    p = Producer({"bootstrap.servers": BOOTSTRAP_SERVERS})
    delivered = []
    def on_delivery(err, msg):
        delivered.append(err)
    p.produce(OUTPUT_TOPIC + "-probe", b"health-check", callback=on_delivery)
    p.flush(timeout=5)
    return len(delivered) > 0 and delivered[0] is None

Wrap this in an HTTP health endpoint on port 8082 and return 200 on success, 503 on failure. Add a Vigilmon monitor at http://<host>:8082/kafka-output-health with the same 1-minute interval.

Step 4: Track Dataflow Processing Throughput

Instrument your Bytewax operators to count items processed and expose a throughput metric:

import time
from collections import deque

# Rolling window of (timestamp, count) tuples — last 60 seconds
_event_times = deque()
_WINDOW_SECONDS = 60

def record_event():
    now = time.time()
    _event_times.append(now)
    # evict old entries
    while _event_times and _event_times[0] < now - _WINDOW_SECONDS:
        _event_times.popleft()

def throughput_per_second():
    return len(_event_times) / _WINDOW_SECONDS

Call record_event() inside your map or filter operator for each processed item. Expose throughput_per_second() via an HTTP endpoint:

# Returns 503 if throughput drops more than 30% below baseline
BASELINE_THROUGHPUT = 1000  # items/sec — set from historical average
THRESHOLD = BASELINE_THROUGHPUT * 0.7

def handle_throughput(path):
    tps = throughput_per_second()
    status = 200 if tps >= THRESHOLD else 503
    return status, {"throughput_per_sec": tps, "threshold": THRESHOLD}

Add a Vigilmon monitor at http://<host>:8083/throughput. Alert on 503 to catch throughput collapses early.

Step 5: Monitor Stateful Operator (SQLite) Health

Bytewax stateful_map and fold_window operators persist state to a SQLite recovery store. A corrupted or write-failed state backend causes the stateful operator to crash.

Add a SQLite health check:

import sqlite3
import os

RECOVERY_DB = "/var/bytewax/recovery.sqlite3"

def check_sqlite_health():
    if not os.path.exists(RECOVERY_DB):
        return False, "state file missing"
    size_mb = os.path.getsize(RECOVERY_DB) / (1024 * 1024)
    try:
        conn = sqlite3.connect(RECOVERY_DB, timeout=3)
        conn.execute("PRAGMA integrity_check").fetchone()
        conn.close()
        return True, {"size_mb": round(size_mb, 2)}
    except sqlite3.Error as e:
        return False, str(e)

Expose this on port 8084 at /state-health, returning 200 when healthy and 503 on any error. Monitor with Vigilmon at 2-minute intervals.

Step 6: Monitor Window Computation and Late Arrival Rate

Late arrivals indicate your watermark is lagging behind real event time, which means windows close incorrectly and downstream consumers see incomplete aggregates.

Track late arrivals in your windowing operator:

_total_events = 0
_late_events = 0

def is_late(event_time, current_watermark):
    return event_time < current_watermark

def record_window_event(event_time, current_watermark):
    global _total_events, _late_events
    _total_events += 1
    if is_late(event_time, current_watermark):
        _late_events += 1

def late_arrival_rate():
    if _total_events == 0:
        return 0.0
    return _late_events / _total_events

Expose /window-health returning 503 when late_arrival_rate() > 0.10 (10% threshold from the spec):

rate = late_arrival_rate()
status = 200 if rate <= 0.10 else 503

Add a Vigilmon monitor on port 8085 at /window-health.

Step 7: Monitor Worker Recovery Health

Bytewax can restart a crashed worker and resume from the SQLite recovery store. Recovery failures cause the dataflow to restart from scratch, losing in-flight state. Instrument recovery attempts:

import logging

_recovery_attempts = 0
_recovery_failures = 0

def on_recovery_attempt():
    global _recovery_attempts
    _recovery_attempts += 1
    logging.info("bytewax_recovery_attempt total=%d", _recovery_attempts)

def on_recovery_result(success: bool):
    global _recovery_failures
    if not success:
        _recovery_failures += 1
        logging.error("bytewax_recovery_failure total=%d", _recovery_failures)

def recovery_health():
    # Return failure if any recovery failed in the last run
    return _recovery_failures == 0

Expose /recovery-health on port 8086. Alert in Vigilmon on any 503 response.

Step 8: Set Up a Heartbeat for the Dataflow Coordinator

When multiple Bytewax workers coordinate via a shared recovery store, track coordinator connectivity with a heartbeat:

  1. In Vigilmon, click Add Monitor → Heartbeat / Cron.
  2. Set the heartbeat interval to 2 minutes.
  3. Copy the heartbeat URL (e.g., https://vigilmon.online/api/v1/heartbeat/<token>).

Ping the heartbeat URL from your dataflow coordination check:

import requests
import threading
import time

HEARTBEAT_URL = "https://vigilmon.online/api/v1/heartbeat/<your-token>"

def heartbeat_loop():
    while True:
        try:
            requests.get(HEARTBEAT_URL, timeout=5)
        except Exception:
            pass
        time.sleep(90)  # ping every 90s, alert fires if missed for 2 min

threading.Thread(target=heartbeat_loop, daemon=True).start()

If the coordinator process stops, the heartbeat goes silent and Vigilmon fires an alert within 2 minutes.

Step 9: Track End-to-End Processing Latency

End-to-end latency is the time from when a Kafka message arrives to when its processed result is written to the output topic. Use a Kafka header to carry the ingestion timestamp:

import time

def add_ingestion_timestamp(record):
    """Attach ingestion timestamp as a field in the record."""
    return {**record, "_ingested_at": time.time()}

def compute_e2e_latency(record):
    """Called at the output operator; returns latency in milliseconds."""
    ingested_at = record.get("_ingested_at")
    if ingested_at:
        return (time.time() - ingested_at) * 1000
    return None

Aggregate p50/p95/p99 latency in a rolling window and expose /latency-health. Return 503 when p99 exceeds your SLA threshold:

P99_SLA_MS = 5000  # 5 second p99 SLA

latencies = []  # rolling list; trim to last 1000

def latency_health():
    if not latencies:
        return 200, {"p99_ms": None}
    sorted_lats = sorted(latencies)
    p99 = sorted_lats[int(len(sorted_lats) * 0.99)]
    status = 200 if p99 <= P99_SLA_MS else 503
    return status, {"p99_ms": p99, "sla_ms": P99_SLA_MS}

Add a Vigilmon monitor on port 8087 at /latency-health.

Step 10: Monitor Python UDF Exception Rate

Unhandled exceptions in Python user-defined functions inside Bytewax operators indicate systematic data quality issues. Wrap UDFs with exception counting:

_udf_calls = 0
_udf_errors = 0

def safe_udf(fn, record):
    global _udf_calls, _udf_errors
    _udf_calls += 1
    try:
        return fn(record)
    except Exception as e:
        _udf_errors += 1
        logging.error("bytewax_udf_error fn=%s error=%s", fn.__name__, e)
        return None  # or re-raise depending on your pipeline's error strategy

def udf_error_rate():
    if _udf_calls == 0:
        return 0.0
    return _udf_errors / _udf_calls

Expose /udf-health returning 503 when udf_error_rate() > 0.01 (1% sustained error rate):

rate = udf_error_rate()
status = 200 if rate <= 0.01 else 503

Monitor on port 8088. A sustained 503 means your UDFs are encountering systematic input data issues.

Alerting Configuration

For each Vigilmon monitor above, configure alerts under the Alerting tab:

| Monitor | Condition | Action | |---------|-----------|--------| | Worker health | Any 503 or connection refused | Page on-call immediately | | Kafka input lag | 503 (lag > threshold) | Page data engineering team | | Kafka output health | Any 503 | Page on-call immediately | | Throughput | 503 (dropped > 30%) | Notify data engineering team | | SQLite state health | Any 503 | Page on-call immediately | | Window / late arrival | 503 (rate > 10%) | Notify data engineering team | | Recovery health | Any 503 | Page on-call immediately | | Coordinator heartbeat | Missed for 2+ minutes | Page on-call immediately | | End-to-end latency | 503 (p99 > SLA) | Notify data engineering team | | UDF exception rate | 503 (rate > 1%) | Notify data engineering team |

Use Vigilmon's Notification Channels to route alerts to Slack, PagerDuty, or email.

Conclusion

Bytewax's Python-native API and Rust runtime make it a powerful choice for production streaming, but that power comes with operational complexity that has no built-in alerting. By adding lightweight sidecar health endpoints to each worker and wiring them into Vigilmon, you get real-time visibility into the ten critical health dimensions of a Bytewax deployment: worker liveness, Kafka I/O health, dataflow throughput, stateful operator integrity, window correctness, worker recovery, coordinator coordination, end-to-end latency, and UDF exception rates. When something breaks — and in streaming, something always eventually breaks — you'll know within a minute instead of after your downstream consumers start complaining.

Get started free at vigilmon.online.

Monitor your app with Vigilmon

Free plan — 5 monitors, no credit card required. Up and running in 60 seconds.

Start free →
← Back to all articles