tutorial

How to Monitor Apache Arrow Flight Services and Data Pipeline Health with Vigilmon

Apache Arrow Flight server failures and in-memory data exchange breakdowns silently stall your high-throughput analytics pipelines. Learn how to monitor Arrow Flight RPC health, data service liveness, and pipeline heartbeats with Vigilmon HTTP probes and heartbeat monitors.

Apache Arrow is the in-memory columnar data format enabling zero-copy data exchange between analytics systems at native CPU speeds — and Apache Arrow Flight is the RPC framework that lets services stream Arrow record batches over gRPC at multi-GB/s throughput. But when a Flight server crashes, a data service fails to serialize an Arrow schema correctly, or a high-memory workload causes a Flight endpoint to OOM, your downstream consumers — Pandas, DuckDB, Spark, ADBC clients — start throwing FlightUnavailableError or hanging indefinitely on stream reads.

Vigilmon gives you external visibility into Arrow Flight service health through HTTP probe monitoring (via a health sidecar alongside your Flight server) and heartbeat monitoring for your Arrow-based data pipelines. This tutorial covers both.

Why Arrow Flight Services Need External Monitoring

Arrow's ecosystem tooling (gRPC reflection, OpenTelemetry traces, custom metrics endpoints) provides runtime visibility — but requires active monitoring infrastructure. External monitoring with Vigilmon adds:

  • Proactive alerting when an Arrow Flight server becomes unreachable (process crash, OOM, port binding failure)
  • Schema and endpoint health verification by checking that the Flight service lists expected endpoints correctly
  • Heartbeat monitoring so you know immediately when a data pipeline that reads or writes Arrow data stops making progress
  • Multi-region availability checking from outside your data plane network

These layers work together: a Flight server can accept gRPC connections while specific endpoints fail due to upstream data source failures, and a data pipeline can be making metadata calls while the actual record batch streaming is stalled on a slow query.

Step 1: Build an Arrow Flight Health Endpoint

Arrow Flight runs over gRPC, which Vigilmon cannot probe directly — you need an HTTP health sidecar alongside your Flight server.

Python Health Sidecar (pyarrow.flight)

# arrow_flight_health.py
import pyarrow.flight as flight
import pyarrow as pa
from flask import Flask, jsonify
import os

app = Flask(__name__)

FLIGHT_HOST = os.environ.get('ARROW_FLIGHT_HOST', 'localhost')
FLIGHT_PORT = int(os.environ.get('ARROW_FLIGHT_PORT', '8815'))

@app.route('/health/arrow-flight')
def flight_health():
    client = None
    try:
        client = flight.connect(f'grpc://{FLIGHT_HOST}:{FLIGHT_PORT}')

        # List available flight endpoints — confirms server is up and responding
        flights = list(client.list_flights())
        endpoint_count = len(flights)

        # Optionally verify a known endpoint is present
        expected_endpoint = os.environ.get('ARROW_FLIGHT_EXPECTED_ENDPOINT')
        if expected_endpoint:
            names = [f.descriptor.path[0].decode() for f in flights
                     if f.descriptor.descriptor_type == flight.DescriptorType.PATH]
            if expected_endpoint not in names:
                return jsonify({
                    'status': 'degraded',
                    'reason': 'expected_endpoint_missing',
                    'endpoint': expected_endpoint,
                    'available': names,
                }), 503

        return jsonify({'status': 'ok', 'endpoints': endpoint_count}), 200

    except Exception as e:
        return jsonify({'status': 'down', 'error': str(e)}), 503
    finally:
        if client:
            try:
                client.close()
            except Exception:
                pass

@app.route('/health/arrow-flight/stream')
def flight_stream_health():
    """Probe a small stream to verify end-to-end record batch delivery."""
    test_endpoint = os.environ.get('ARROW_FLIGHT_TEST_ENDPOINT')
    if not test_endpoint:
        return jsonify({'status': 'skipped', 'reason': 'no_test_endpoint_configured'}), 200

    client = None
    try:
        client = flight.connect(f'grpc://{FLIGHT_HOST}:{FLIGHT_PORT}')
        descriptor = flight.FlightDescriptor.for_path(test_endpoint)
        info = client.get_flight_info(descriptor)

        if not info.endpoints:
            return jsonify({'status': 'degraded', 'reason': 'no_endpoints_in_flight_info'}), 503

        # Read just the first batch to verify streaming works
        reader = client.do_get(info.endpoints[0].ticket)
        first_batch = reader.read_chunk()

        return jsonify({
            'status': 'ok',
            'schema_fields': len(first_batch.data.schema),
            'rows_in_first_batch': first_batch.data.num_rows,
        }), 200

    except Exception as e:
        return jsonify({'status': 'down', 'error': str(e)}), 503
    finally:
        if client:
            try:
                client.close()
            except Exception:
                pass

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

Java/Scala Health Sidecar (Arrow Flight Java SDK)

// ArrowFlightHealthController.java
@RestController
public class ArrowFlightHealthController {

    private final String flightHost = System.getenv("ARROW_FLIGHT_HOST");
    private final int flightPort = Integer.parseInt(
        System.getenv().getOrDefault("ARROW_FLIGHT_PORT", "8815")
    );

    @GetMapping("/health/arrow-flight")
    public ResponseEntity<Map<String, Object>> checkHealth() {
        Map<String, Object> status = new LinkedHashMap<>();

        try (FlightClient client = FlightClient.builder()
                .location(Location.forGrpcInsecure(flightHost, flightPort))
                .build()) {

            // List flights — confirms gRPC connectivity and server responsiveness
            Iterable<FlightInfo> flights = client.listFlights(Criteria.ALL);
            List<String> endpointNames = new ArrayList<>();
            for (FlightInfo info : flights) {
                if (info.getDescriptor().getType() == FlightDescriptor.DescriptorType.PATH) {
                    endpointNames.add(info.getDescriptor().getPath().get(0));
                }
            }

            status.put("status", "ok");
            status.put("endpoints", endpointNames.size());
            return ResponseEntity.ok(status);

        } catch (Exception e) {
            status.put("status", "down");
            status.put("error", e.getMessage());
            return ResponseEntity.status(503).body(status);
        }
    }
}

Node.js Health Sidecar (via HTTP/gRPC-Web gateway)

If your Arrow Flight server sits behind an Envoy gRPC-Web proxy:

// arrow-health.js
const express = require('express');
const axios = require('axios');

const app = express();

const GRPC_WEB_GATEWAY = process.env.ARROW_GRPC_WEB_URL || 'http://localhost:8816';

app.get('/health/arrow-flight', async (req, res) => {
  try {
    // Probe the gRPC-Web health check endpoint (grpc.health.v1.Health)
    const r = await axios.get(`${GRPC_WEB_GATEWAY}/healthz`, { timeout: 6000 });
    if (r.status === 200) {
      return res.status(200).json({ status: 'ok' });
    }
    return res.status(503).json({ status: 'degraded', code: r.status });
  } catch (err) {
    return res.status(503).json({ status: 'down', error: err.message });
  }
});

app.listen(3009);

Step 2: Configure Vigilmon HTTP Monitor for Arrow Flight

  1. Log in to vigilmon.online and go to Monitors → New Monitor
  2. Choose HTTP / HTTPS
  3. Set the URL to your Arrow Flight health endpoint: https://your-app.example.com/health/arrow-flight
  4. Set the check interval to 1 minute
  5. Under Expected response, configure:
    • Status code: 200
    • Response body contains: "status":"ok"
    • Response time threshold: 5000ms (gRPC list_flights involves schema metadata reads)
  6. Under Alert channels, assign your Slack or PagerDuty channel
  7. Save the monitor

Add a second monitor for stream health (if your use case warrants it):

  • URL: https://your-app.example.com/health/arrow-flight/stream
  • Expected: 200, body contains "status":"ok"
  • Interval: 5 minutes (streaming probes are more expensive)
  • Alert channel: data-platform on-call channel

Vigilmon's multi-region probe consensus prevents false alerts from transient gRPC connection setup latency.

Step 3: Heartbeat Monitoring for Arrow Data Pipelines

Arrow Flight service health is necessary — but not sufficient. Your data pipelines that ingest, transform, or export Arrow record batches can be stalled on slow upstream queries, blocking on memory allocation failures, or silently emitting empty record batches due to filter logic bugs — while the Flight server itself looks healthy.

Vigilmon heartbeat monitors detect silent pipeline stalls: your pipeline pings Vigilmon after each successful batch of record batch processing. If pings stop, Vigilmon alerts.

Set Up the Heartbeat Monitor

  1. In Vigilmon, go to Monitors → New Monitor → Heartbeat
  2. Set the name: arrow-data-pipeline
  3. Set the expected interval: 15 minutes (or your pipeline's processing cadence)
  4. Set the grace period: 5 minutes
  5. Save — copy the heartbeat URL, e.g. https://vigilmon.online/heartbeat/abc123xyz

Python Arrow Pipeline with Heartbeat

# arrow_pipeline.py
import pyarrow as pa
import pyarrow.flight as flight
import pyarrow.parquet as pq
import requests, os

FLIGHT_HOST = os.environ['ARROW_FLIGHT_HOST']
FLIGHT_PORT = int(os.environ.get('ARROW_FLIGHT_PORT', '8815'))
VIGILMON_HB = os.environ['VIGILMON_HEARTBEAT_URL']

def process_pipeline():
    client = flight.connect(f'grpc://{FLIGHT_HOST}:{FLIGHT_PORT}')

    # Read data from Arrow Flight server
    descriptor = flight.FlightDescriptor.for_path('daily_events')
    info = client.get_flight_info(descriptor)

    batches_processed = 0
    rows_total = 0

    for endpoint in info.endpoints:
        reader = client.do_get(endpoint.ticket)
        for chunk in reader:
            batch = chunk.data

            # Apply transformations using Arrow compute kernels
            transformed = transform_batch(batch)

            # Write to Parquet or push to downstream Flight server
            write_batch(transformed)

            batches_processed += 1
            rows_total += batch.num_rows

    client.close()
    print(f"Processed {batches_processed} batches, {rows_total} rows")
    return rows_total

def transform_batch(batch):
    import pyarrow.compute as pc
    # Example: filter and project
    mask = pc.greater(batch.column('value'), pa.scalar(0))
    return batch.filter(mask)

def write_batch(batch):
    # Write to your downstream target
    pass

if __name__ == '__main__':
    rows = process_pipeline()
    if rows > 0:
        # Only heartbeat on non-empty successful runs
        requests.get(VIGILMON_HB, timeout=5)
        print("Vigilmon heartbeat sent.")
    else:
        print("WARNING: Zero rows processed — NOT sending heartbeat.")

Pandas/PyArrow ETL with Heartbeat

# pandas_arrow_etl.py
import pandas as pd
import pyarrow as pa
import pyarrow.dataset as ds
import requests, os

def run_etl():
    # Read from Parquet dataset using Arrow
    dataset = ds.dataset(
        os.environ['INPUT_PATH'],
        format='parquet',
        partitioning='hive'
    )

    # Scan and filter using Arrow push-down predicates
    table = dataset.to_table(
        filter=ds.field('date') == ds.scalar(os.environ['PROCESS_DATE']),
        columns=['id', 'value', 'category']
    )

    if table.num_rows == 0:
        raise ValueError(f"No rows found for date {os.environ['PROCESS_DATE']}")

    # Transform
    df = table.to_pandas()
    df['value_normalized'] = df['value'] / df['value'].max()
    result_table = pa.Table.from_pandas(df)

    # Write output
    import pyarrow.parquet as pq
    pq.write_table(result_table, os.environ['OUTPUT_PATH'])

    return table.num_rows

rows = run_etl()
print(f"ETL complete: {rows} rows")
requests.get(os.environ['VIGILMON_HEARTBEAT_URL'], timeout=5)

Rust Arrow Pipeline with Heartbeat (datafusion)

// main.rs
use datafusion::prelude::*;
use reqwest::blocking::get;
use std::env;

#[tokio::main]
async fn main() -> datafusion::error::Result<()> {
    let ctx = SessionContext::new();

    // Register Parquet data source
    ctx.register_parquet(
        "events",
        &env::var("INPUT_PATH").unwrap(),
        ParquetReadOptions::default(),
    ).await?;

    // Run transformation query using Arrow/DataFusion
    let df = ctx.sql(
        "SELECT category, SUM(value) as total \
         FROM events \
         WHERE date = '2026-07-06' \
         GROUP BY category"
    ).await?;

    let batches = df.collect().await?;
    let row_count: usize = batches.iter().map(|b| b.num_rows()).sum();
    println!("Processed {} rows", row_count);

    // Ping Vigilmon heartbeat on success
    if row_count > 0 {
        if let Ok(hb_url) = env::var("VIGILMON_HEARTBEAT_URL") {
            let _ = get(&hb_url);
        }
    }

    Ok(())
}

Step 4: Alert Routing for Arrow Flight Failures

Arrow Flight failures tend to be immediate and complete: when the gRPC server crashes, all consumers get FlightUnavailableError immediately. Data pipeline failures are more gradual — silent empty outputs, OOM-caused restarts, slow-growing memory pressure.

Configure alert routing in Vigilmon:

| Monitor | Alert Channel | Priority | |---|---|---| | Arrow Flight health /health/arrow-flight | Slack + PagerDuty | P1 | | Arrow Flight stream probe /health/arrow-flight/stream | Slack | P2 | | Heartbeat: data ingestion pipeline | Slack + email | P1 | | Heartbeat: transformation pipeline | Slack | P2 | | Heartbeat: export/delivery pipeline | Email | P3 |

Set response time thresholds for early warning:

  • Alert at 3000ms for the Flight health endpoint (slow gRPC list_flights signals server pressure or memory pressure)
  • Alert at 10000ms for the stream probe (slow record batch delivery signals upstream query performance degradation)

For memory-intensive Arrow workloads, configure a tight heartbeat interval with no grace period for the ingestion pipeline — Arrow OOM failures are often fast and total, with no graceful degradation before complete failure.

Summary

Arrow Flight service failures are usually fast and complete, while data pipeline failures are often silent and gradual. External monitoring covers both:

| Monitor Type | What It Covers | |---|---| | HTTP monitor on /health/arrow-flight | gRPC server availability, endpoint registration | | HTTP monitor on /health/arrow-flight/stream | End-to-end record batch streaming health | | Heartbeat monitor | Pipeline liveness, data processing progress |

Get started free at vigilmon.online — your first Arrow Flight 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