tutorial

How to Monitor Apache Beam Pipelines with Vigilmon

Apache Beam pipeline failures are silent — element count lag, stalled watermarks, and runner worker crashes don't surface as HTTP errors. Learn how to monitor Beam pipelines running on Dataflow, Flink, or Spark with Vigilmon external checks and heartbeat monitoring.

Apache Beam pipelines fail quietly. A watermark that stops advancing, a worker that crashes mid-bundle, or a source that falls behind all look identical from the outside — the job appears to be running while elements pile up unprocessed. Standard runner dashboards (Dataflow monitoring, Flink Web UI, Spark History Server) give you rich internal metrics, but only if someone is watching. External monitoring with Vigilmon gives you proactive alerting the moment pipeline health degrades.

This tutorial covers building HTTP health endpoints for Beam pipelines and wiring them into Vigilmon monitors, with examples for both batch and streaming use cases.

Why Beam Pipelines Need External Monitoring

Apache Beam's execution model abstracts over runners — the same pipeline code runs on Dataflow, Flink, Spark, or the direct runner. Each runner exposes different internal metrics, which makes unified monitoring difficult. Vigilmon fills the gap:

  • Runner-agnostic SLA monitoring — a single Vigilmon HTTP check works regardless of which runner your pipeline targets
  • Watermark stall detection — streaming pipelines can freeze at a specific timestamp without any error; a health sidecar that exposes watermark progress makes this visible
  • Worker failure rate tracking — Dataflow and Flink expose worker error counts via their REST APIs; a health endpoint surfaces this as a 503 before it breaches SLA
  • End-to-end throughput heartbeats — a sink sidecar that pings Vigilmon after every successful output batch proves the pipeline is actually producing, not just running

Step 1: Build a Pipeline Health Endpoint

Beam runners expose health data through different APIs. Build a thin HTTP sidecar that normalizes this into a standard health endpoint.

Dataflow Runner Health Sidecar (Python)

# health_server.py
import os
import time
from flask import Flask, jsonify
from googleapiclient import discovery

app = Flask(__name__)
project = os.environ['GCP_PROJECT']
region = os.environ['DATAFLOW_REGION']
job_id = os.environ['DATAFLOW_JOB_ID']

def get_job_health():
    service = discovery.build('dataflow', 'v1b3')
    job = service.projects().locations().jobs().get(
        projectId=project,
        location=region,
        jobId=job_id,
        view='JOB_VIEW_ALL'
    ).execute()

    state = job.get('currentState', 'UNKNOWN')
    # Running states: JOB_STATE_RUNNING, JOB_STATE_PENDING
    if state not in ('JOB_STATE_RUNNING', 'JOB_STATE_PENDING'):
        return False, {'state': state, 'reason': 'job_not_running'}

    # Check for worker errors in the last 5 minutes
    metrics = service.projects().locations().jobs().getMetrics(
        projectId=project, location=region, jobId=job_id
    ).execute()

    worker_errors = 0
    for metric in metrics.get('metrics', []):
        if metric.get('name', {}).get('name') == 'WorkerErrors':
            worker_errors = int(metric.get('scalar', 0))

    if worker_errors > int(os.environ.get('MAX_WORKER_ERRORS', '5')):
        return False, {'state': state, 'worker_errors': worker_errors}

    return True, {'state': state, 'worker_errors': worker_errors}

@app.route('/health/beam')
def health():
    healthy, details = get_job_health()
    status = 'ok' if healthy else 'degraded'
    code = 200 if healthy else 503
    return jsonify({'status': status, **details}), code

if __name__ == '__main__':
    app.run(port=8090)

Flink Runner Health Sidecar (Java)

// BeamHealthEndpoint.java
import com.sun.net.httpserver.HttpServer;
import java.net.URI;
import java.net.http.HttpClient;
import java.net.http.HttpRequest;
import java.net.http.HttpResponse;

public class BeamHealthEndpoint {
    static final String FLINK_API = System.getenv("FLINK_REST_URL"); // e.g. http://jobmanager:8081
    static final String JOB_ID   = System.getenv("FLINK_JOB_ID");
    static final HttpClient http  = HttpClient.newHttpClient();

    public static void main(String[] args) throws Exception {
        HttpServer server = HttpServer.create(new java.net.InetSocketAddress(8090), 0);
        server.createContext("/health/beam", exchange -> {
            String result = checkFlinkJob();
            byte[] body = result.getBytes();
            int code = result.contains("\"ok\"") ? 200 : 503;
            exchange.sendResponseHeaders(code, body.length);
            exchange.getResponseBody().write(body);
            exchange.close();
        });
        server.start();
    }

    static String checkFlinkJob() {
        try {
            HttpResponse<String> resp = http.send(
                HttpRequest.newBuilder(URI.create(FLINK_API + "/jobs/" + JOB_ID)).GET().build(),
                HttpResponse.BodyHandlers.ofString()
            );
            if (resp.statusCode() != 200) return "{\"status\":\"down\",\"reason\":\"flink_api_error\"}";

            // Check job state and checkpointing lag
            String body = resp.body();
            boolean running = body.contains("\"RUNNING\"");
            return running
                ? "{\"status\":\"ok\"}"
                : "{\"status\":\"degraded\",\"reason\":\"not_running\",\"detail\":" + body + "}";
        } catch (Exception e) {
            return "{\"status\":\"down\",\"error\":\"" + e.getMessage() + "\"}";
        }
    }
}

Watermark Progress Endpoint (Streaming Pipelines)

For streaming Beam pipelines, watermark stalls are the most common silent failure. Expose watermark lag from your pipeline's metrics:

# watermark_health.py — add to the Flask app above
import apache_beam as beam
from apache_beam.runners.dataflow.dataflow_runner import DataflowRunner

@app.route('/health/beam/watermark')
def watermark_health():
    # Read watermark lag from your pipeline's published metric
    # This assumes you publish a custom metric in your DoFn:
    # beam.metrics.Metrics.gauge('pipeline', 'watermark_lag_seconds').set(lag)
    service = discovery.build('dataflow', 'v1b3')
    metrics = service.projects().locations().jobs().getMetrics(
        projectId=project, location=region, jobId=job_id
    ).execute()

    watermark_lag = None
    for metric in metrics.get('metrics', []):
        if metric.get('name', {}).get('name') == 'watermark_lag_seconds':
            watermark_lag = int(metric.get('scalar', 0))

    if watermark_lag is None:
        return jsonify({'status': 'unknown', 'reason': 'metric_not_found'}), 200

    threshold = int(os.environ.get('MAX_WATERMARK_LAG_SECONDS', '300'))
    if watermark_lag > threshold:
        return jsonify({
            'status': 'degraded',
            'reason': 'watermark_stalled',
            'lag_seconds': watermark_lag,
            'threshold': threshold,
        }), 503

    return jsonify({'status': 'ok', 'lag_seconds': watermark_lag}), 200

Publish the watermark lag metric inside your streaming DoFn:

class ProcessElementFn(beam.DoFn):
    watermark_lag = beam.metrics.Metrics.gauge('pipeline', 'watermark_lag_seconds')

    def process(self, element, timestamp=beam.DoFn.TimestampParam):
        import time
        lag = int(time.time() - float(timestamp))
        self.watermark_lag.set(lag)
        yield element

Step 2: Configure Vigilmon HTTP Monitors

  1. Log in to vigilmon.online and go to Monitors → New Monitor
  2. Choose HTTP / HTTPS
  3. Set the URL to your pipeline health sidecar: https://your-dataflow-sidecar.example.com/health/beam
  4. Set the check interval to 2 minutes
  5. Under Expected response:
    • Status code: 200
    • Response body contains: "status":"ok"
    • Response time threshold: 10000ms
  6. Under Alert channels, assign your on-call Slack channel or PagerDuty integration
  7. Save the monitor

Add a second monitor for watermark lag on streaming pipelines:

  • URL: https://your-dataflow-sidecar.example.com/health/beam/watermark
  • Expected: 200, body contains "status":"ok"
  • Interval: 2 minutes
  • Alert channel: separate Slack channel for data freshness alerts

Step 3: Heartbeat Monitoring for Pipeline Output

A pipeline can be running and processing elements while the output sink is silently failing — writes to BigQuery, GCS, or a downstream Kafka topic fail and elements are dropped. Vigilmon heartbeat monitors detect this by requiring your pipeline's sink to ping Vigilmon after each successful write batch.

Set Up the Heartbeat Monitor

  1. In Vigilmon, go to Monitors → New Monitor → Heartbeat
  2. Name it: beam-orders-pipeline-sink
  3. Set the expected interval to 10 minutes (adjust to your pipeline's output batch size)
  4. Set the grace period: 5 minutes
  5. Save and copy the heartbeat URL: https://vigilmon.online/heartbeat/abc123xyz

Wire Heartbeats Into Your Sink DoFn

# sink_fn.py
import requests, os
import apache_beam as beam

class SinkWithHeartbeat(beam.DoFn):
    def __init__(self, heartbeat_url):
        self.heartbeat_url = heartbeat_url
        self._batch_count = 0

    def process(self, element):
        write_to_bigquery(element)
        self._batch_count += 1
        # Ping Vigilmon every 500 elements
        if self._batch_count % 500 == 0:
            try:
                requests.get(self.heartbeat_url, timeout=3)
            except Exception:
                pass  # Don't fail the pipeline on heartbeat errors
        yield element

# In your pipeline:
(
    pipeline
    | 'ReadFromSource' >> beam.io.ReadFromPubSub(subscription=SUBSCRIPTION)
    | 'ProcessElements' >> beam.ParDo(ProcessElementFn())
    | 'WriteToSink' >> beam.ParDo(SinkWithHeartbeat(os.environ['VIGILMON_HEARTBEAT_URL']))
)

For batch pipelines where the job completes and exits, send the heartbeat from the pipeline's final step instead of on an interval:

class FinalHeartbeatFn(beam.DoFn):
    def finish_bundle(self):
        url = os.environ.get('VIGILMON_HEARTBEAT_URL')
        if url:
            requests.get(url, timeout=5)

pipeline | 'FinalStep' >> beam.ParDo(FinalHeartbeatFn())

Step 4: Alert Routing for Pipeline Failures

Configure alert severity to match pipeline criticality:

| Monitor | Alert Channel | Priority | |---|---|---| | Beam runner health /health/beam | Slack + PagerDuty | P1 | | Watermark lag /health/beam/watermark | Slack | P2 | | Heartbeat: pipeline sink | Slack + email | P1 | | Batch job heartbeat | Email | P2 |

Set response time thresholds on the runner health endpoint — Dataflow API latency above 15 seconds often precedes worker scaling failures, giving you an early warning before elements actually stop processing.

Summary

Apache Beam pipelines need monitoring at three layers: the runner, the watermark, and the output sink.

| Monitor Type | What It Covers | |---|---| | HTTP monitor on /health/beam | Runner job state, worker error count | | HTTP monitor on /health/beam/watermark | Streaming watermark stall, data freshness | | Heartbeat monitor on sink | End-to-end pipeline liveness, output health |

Get started free at vigilmon.online — your first Beam pipeline monitor is running in under two minutes.

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