How to Monitor Apache ORC File Processing Health with Vigilmon

Apache ORC (Optimized Row Columnar) is a high-performance columnar storage format widely used in Hadoop, Hive, Spark, and Presto data pipelines. ORC files are compact and fast — but ORC processing pipelines fail in quiet ways: a writer stalls with no bytes flushed, a Hive metastore falls out of sync with the physical files, or a Spark job reads corrupt ORC stripes and silently skips rows without raising an exception.

Vigilmon gives you external visibility into ORC pipeline health through HTTP probe monitoring for your batch job orchestrators and heartbeat monitoring for your ORC writer processes. This tutorial covers both.

Why ORC Pipelines Need External Monitoring

ORC processing is typically embedded in Spark, Hive, or Presto jobs managed by orchestrators like Airflow or Oozie. Failures look like:

• Stalled writers — a Spark executor hangs while writing an ORC file, the job never completes, and downstream queries return stale data without error
• Schema drift — a new column appears upstream, ORC reader silently ignores it (or panics), and downstream aggregations are subtly wrong
• Corrupt stripe detection — a failed mid-write leaves an ORC file with a valid header but corrupt stripes, which some readers will skip silently
• Zero-byte files — an error causes a Spark partition to emit an empty ORC file, which passes validation but produces incorrect query results

External monitoring with Vigilmon adds:

• Proactive alerting when ORC batch jobs exceed their SLA window
• Writer liveness detection via heartbeat monitors on long-running ORC writer processes
• Row count anomaly alerts via a health endpoint that validates output file stats
• Pipeline stall detection when no new ORC files land in a partition for longer than expected

Step 1: Build ORC Pipeline Health Endpoints

ORC itself has no built-in HTTP API. You expose health through your job orchestrator (Airflow, Oozie) or a custom sidecar on your Spark driver.

Airflow ORC Job Health Endpoint

# orc_health_api.py — expose via Flask alongside your Airflow instance
from flask import Flask, jsonify
from airflow.models import DagRun, TaskInstance
from airflow.utils.state import State
from datetime import datetime, timedelta
import os

app = Flask(__name__)

ORC_DAG_ID = os.environ.get('ORC_DAG_ID', 'orc_etl_pipeline')
SLA_MINUTES = int(os.environ.get('ORC_SLA_MINUTES', '60'))

@app.route('/health/orc-pipeline')
def orc_pipeline_health():
    try:
        from airflow.utils.session import create_session
        with create_session() as session:
            # Check most recent DAG run
            latest_run = session.query(DagRun).filter(
                DagRun.dag_id == ORC_DAG_ID
            ).order_by(DagRun.execution_date.desc()).first()

            if not latest_run:
                return jsonify({'status': 'degraded', 'reason': 'no_runs_found'}), 503

            age_minutes = (datetime.utcnow() - latest_run.execution_date).total_seconds() / 60

            if latest_run.state == State.FAILED:
                return jsonify({
                    'status': 'down',
                    'reason': 'last_run_failed',
                    'run_id': latest_run.run_id,
                    'execution_date': latest_run.execution_date.isoformat(),
                }), 503

            if latest_run.state == State.RUNNING and age_minutes > SLA_MINUTES:
                return jsonify({
                    'status': 'degraded',
                    'reason': 'run_exceeded_sla',
                    'age_minutes': round(age_minutes, 1),
                    'sla_minutes': SLA_MINUTES,
                }), 503

            if latest_run.state == State.SUCCESS and age_minutes > SLA_MINUTES * 2:
                return jsonify({
                    'status': 'degraded',
                    'reason': 'no_recent_successful_run',
                    'last_success_minutes_ago': round(age_minutes, 1),
                }), 503

            return jsonify({
                'status': 'ok',
                'last_run_state': latest_run.state,
                'last_run_minutes_ago': round(age_minutes, 1),
            })
    except Exception as e:
        return jsonify({'status': 'down', 'error': str(e)}), 503

if __name__ == '__main__':
    app.run(host='0.0.0.0', port=3007)

ORC Output Validation Endpoint

This endpoint checks that ORC files are actually landing in HDFS or S3 with non-zero row counts:

@app.route('/health/orc-output')
def orc_output_health():
    import subprocess, json

    output_path = os.environ.get('ORC_OUTPUT_PATH', 'hdfs:///data/orc/events')
    partition_format = os.environ.get('ORC_PARTITION_FORMAT', 'dt=%Y-%m-%d')
    today_partition = datetime.utcnow().strftime(partition_format)
    full_path = f'{output_path}/{today_partition}'

    try:
        # Check HDFS for today's ORC files
        result = subprocess.run(
            ['hdfs', 'dfs', '-ls', full_path],
            capture_output=True, text=True, timeout=10
        )

        if result.returncode != 0:
            return jsonify({
                'status': 'degraded',
                'reason': 'partition_not_found',
                'path': full_path,
            }), 503

        lines = [l for l in result.stdout.splitlines() if l.strip() and not l.startswith('Found')]
        if not lines:
            return jsonify({
                'status': 'degraded',
                'reason': 'no_orc_files_in_partition',
                'path': full_path,
            }), 503

        # Check for zero-byte files
        zero_byte = [l for l in lines if l.split()[4] == '0']
        if zero_byte:
            return jsonify({
                'status': 'degraded',
                'reason': 'zero_byte_orc_files_detected',
                'count': len(zero_byte),
            }), 503

        return jsonify({
            'status': 'ok',
            'partition': today_partition,
            'orc_file_count': len(lines),
        })
    except subprocess.TimeoutExpired:
        return jsonify({'status': 'degraded', 'reason': 'hdfs_ls_timeout'}), 503
    except Exception as e:
        return jsonify({'status': 'down', 'error': str(e)}), 503

Spark ORC Writer Health (PySpark)

# orc_writer_job.py
from pyspark.sql import SparkSession
import requests, os, time

spark = SparkSession.builder.appName('orc-writer').getOrCreate()
heartbeat_url = os.environ.get('VIGILMON_HEARTBEAT_URL')

def write_orc_partition(df, output_path, partition_col):
    df.write.partitionBy(partition_col).mode('overwrite').orc(output_path)

    # Validate output before pinging heartbeat
    written = spark.read.orc(output_path)
    row_count = written.count()

    if row_count == 0:
        raise ValueError(f'ORC write produced zero rows at {output_path}')

    # Ping Vigilmon: ORC write completed with valid output
    if heartbeat_url:
        requests.get(heartbeat_url, timeout=5)

    return row_count

# Main ETL loop
while True:
    try:
        df = extract_from_source()
        count = write_orc_partition(df, os.environ['ORC_OUTPUT_PATH'], 'dt')
        print(f'Wrote {count} rows to ORC')
    except Exception as e:
        print(f'ORC write failed: {e}')
        # No heartbeat ping — Vigilmon will fire alert after grace period
    time.sleep(int(os.environ.get('WRITE_INTERVAL_SECONDS', '300')))

Step 2: Configure Vigilmon HTTP Monitor for ORC Pipelines

1. Log in to vigilmon.online and go to Monitors → New Monitor
2. Choose HTTP / HTTPS
3. Set the URL to your ORC pipeline health endpoint: https://your-app.example.com/health/orc-pipeline
4. Set the check interval to 5 minutes (ORC batch jobs run less frequently than real-time services)
5. Under Expected response, configure:
   • Status code: 200
   • Response body contains: "status":"ok"
   • Response time threshold: 10000ms (Airflow API queries can be slow)
6. Under Alert channels, assign your Slack or PagerDuty channel
7. Save the monitor

Add a second monitor for ORC output validation:

• URL: https://your-app.example.com/health/orc-output
• Expected: 200, body contains "status":"ok"
• Interval: 15 minutes (partition output check)
• Alert channel: data engineering Slack channel

Step 3: Heartbeat Monitoring for ORC Writer Processes

Long-running ORC writer jobs — Spark streaming jobs writing micro-batches, Flink ORC sinks, or custom writers — need heartbeat monitoring to detect stalls that don't surface as job failures.

Set Up the Heartbeat Monitor

1. In Vigilmon, go to Monitors → New Monitor → Heartbeat
2. Set the name: orc-events-writer
3. Set the expected interval: 10 minutes (match your ORC write batch frequency)
4. Set the grace period: 20 minutes
5. Save — copy the heartbeat URL, e.g. https://vigilmon.online/heartbeat/abc123xyz

Wire It Into a Flink ORC Sink (Java)

// OrcWriterHeartbeat.java
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
import java.net.http.*;
import java.net.URI;

public class OrcWriterHeartbeat<T> extends RichSinkFunction<T> {
    private final String vigilmonUrl;
    private final int batchSize;
    private int count = 0;
    private transient HttpClient http;

    public OrcWriterHeartbeat(String vigilmonUrl, int batchSize) {
        this.vigilmonUrl = vigilmonUrl;
        this.batchSize = batchSize;
    }

    @Override
    public void open(org.apache.flink.configuration.Configuration params) {
        http = HttpClient.newHttpClient();
    }

    @Override
    public void invoke(T value, Context ctx) throws Exception {
        writeToOrc(value);
        count++;
        if (count % batchSize == 0) {
            // Ping after each committed batch
            http.sendAsync(
                HttpRequest.newBuilder(URI.create(vigilmonUrl)).GET().build(),
                HttpResponse.BodyHandlers.discarding()
            );
        }
    }

    private void writeToOrc(T value) {
        // ... actual ORC write logic ...
    }
}

Wire It Into a Batch Script

#!/bin/bash
# run_orc_job.sh
set -e

HEARTBEAT_URL="${VIGILMON_HEARTBEAT_URL}"

echo "Starting ORC batch write..."
spark-submit \
  --class com.example.OrcEtlJob \
  --master yarn \
  orc-etl-job.jar

echo "ORC batch write completed successfully"

# Ping Vigilmon only on success — failure exits above
if [ -n "$HEARTBEAT_URL" ]; then
  curl -sf "$HEARTBEAT_URL" > /dev/null
fi

Schedule via cron every 30 minutes. Vigilmon alerts if the heartbeat stops arriving.

Step 4: Alert Routing for ORC Pipeline Failures

ORC failures are data quality failures — they produce stale, incomplete, or subtly wrong query results. Route alerts with that context:

| Monitor | Alert Channel | Priority | |---|---|---| | ORC pipeline /health/orc-pipeline | Slack + email | P2 | | ORC output /health/orc-output | Slack | P2 | | Heartbeat: Spark ORC writer | Slack + PagerDuty | P1 (if SLA-critical) | | Heartbeat: Flink ORC sink | Slack | P2 |

Set response time thresholds appropriate for batch workloads:

• Alert at 8000ms for the pipeline health endpoint (Airflow API slowness signals scheduler overload)
• Alert at 15000ms for HDFS output checks (slow NameNode responses signal cluster pressure)

For SLA-critical ORC pipelines (e.g., end-of-day reporting), add a Vigilmon status page entry so stakeholders can self-serve on pipeline health during incidents.

Summary

ORC pipeline failures are data quality failures — they don't throw visible errors, they produce wrong or missing results. External monitoring catches stalls, zero-byte outputs, and stalled writers before downstream consumers notice:

| Monitor Type | What It Covers | |---|---| | HTTP monitor on /health/orc-pipeline | DAG run state, SLA compliance, last run age | | HTTP monitor on /health/orc-output | Partition presence, file count, zero-byte detection | | Heartbeat monitor | ORC writer process liveness, batch completion confirmation |

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