tutorial

How to Monitor Your Apache Parquet Data Pipelines (Free, Multi-Region)

Apache Parquet is the de-facto columnar format for big data — but corrupt files and schema evolution failures fail silently in batch pipelines. Learn how to add external uptime monitoring, heartbeat checks for ETL jobs, and instant alerts — free.

How to Monitor Your Apache Parquet Data Pipelines (Free, Multi-Region)

Apache Parquet is the columnar storage format that underpins most modern data lake architectures. Spark, Flink, Presto, Hive, Pandas, and DuckDB all read and write Parquet natively. Because it compresses so well and serves analytical queries so efficiently, it has become the default format for landing data in S3, GCS, ADLS, and HDFS. But Parquet's reliability is deceptive — a partially written file from a crashed writer is a valid file header with a corrupted footer, not an obvious error. Schema evolution without proper metadata management causes silent column drops. Row group statistics that drift out of sync cause predicate pushdown to skip correct rows.

By the end of this guide you'll have external uptime monitoring, multi-region health probes, heartbeat monitoring for Parquet ETL pipelines, and instant alerts — all on the free tier.

Why Parquet pipeline failures are hard to catch

Parquet failures manifest as data quality problems rather than system errors:

Corrupt file footers — Parquet files store their metadata in the footer (last bytes of the file). A writer that crashes mid-flush produces a file with a valid magic number but a truncated or missing footer. PyArrow and pandas will raise ArrowInvalid when reading it, but only when that specific partition is accessed — not at write time.

Schema evolution mismatches — When a producer adds, renames, or changes the type of a column without coordinating with consumers, downstream jobs silently produce null for that column in all rows. No exception is raised; the column just reads as all nulls.

Statistics corruption — Parquet row groups embed min/max statistics for each column to enable predicate pushdown. Writers that crash or are misconfigured can write incorrect statistics, causing query engines to skip row groups they should read. Queries return silently incomplete results.

Small file proliferation — Streaming writers (Kafka Connect, Flink) writing one file per partition per micro-batch can create millions of tiny Parquet files. Read performance degrades catastrophically, but no error is raised — queries just run 10× slower.

Step 1: Expose a health endpoint that validates Parquet read/write

A health endpoint must exercise real Parquet I/O to prove your pipeline infrastructure is healthy:

With FastAPI (Python):

import io
import os
import pyarrow as pa
import pyarrow.parquet as pq
from fastapi import FastAPI
from fastapi.responses import JSONResponse

app = FastAPI()

@app.get("/health")
def health():
    try:
        start = __import__('time').time()

        # Write a probe Parquet file in memory
        schema = pa.schema([
            pa.field("probe_id", pa.int64()),
            pa.field("probe_ts", pa.int64()),
            pa.field("probe_value", pa.float64()),
            pa.field("probe_label", pa.string()),
        ])
        table = pa.table({
            "probe_id": [1, 2, 3],
            "probe_ts": [1000000, 1000001, 1000002],
            "probe_value": [1.1, 2.2, 3.3],
            "probe_label": ["a", "b", "c"],
        }, schema=schema)

        buf = io.BytesIO()
        pq.write_table(table, buf, compression="snappy")

        # Read it back and verify schema and row count
        buf.seek(0)
        pf = pq.ParquetFile(buf)
        meta = pf.metadata
        read_table = pf.read()

        if read_table.num_rows != 3:
            raise ValueError(f"Expected 3 rows, got {read_table.num_rows}")
        if set(read_table.column_names) != {"probe_id", "probe_ts", "probe_value", "probe_label"}:
            raise ValueError(f"Schema mismatch: {read_table.column_names}")

        elapsed_ms = (time.time() - start) * 1000
        return {
            "status": "UP",
            "parquet_version": meta.format_version,
            "rows_verified": read_table.num_rows,
            "round_trip_ms": round(elapsed_ms, 2),
            "compression": "snappy",
        }
    except Exception as e:
        return JSONResponse(status_code=500, content={"status": "DOWN", "error": str(e)})

With Flask (Python):

import io
import time
import pyarrow as pa
import pyarrow.parquet as pq
from flask import Flask, jsonify

app = Flask(__name__)

@app.get("/health")
def health():
    try:
        start = time.time()
        schema = pa.schema([
            pa.field("id", pa.int64()),
            pa.field("value", pa.float64()),
        ])
        table = pa.table({"id": [1, 2], "value": [1.0, 2.0]}, schema=schema)
        buf = io.BytesIO()
        pq.write_table(table, buf)
        buf.seek(0)
        rt = pq.read_table(buf)
        assert rt.num_rows == 2
        return jsonify({
            "status": "UP",
            "rows": rt.num_rows,
            "ms": round((time.time() - start) * 1000, 2),
        })
    except Exception as e:
        return jsonify({"status": "DOWN", "error": str(e)}), 500

The health check validates the full Parquet write-read round-trip including schema enforcement and row count. A PyArrow version incompatibility, a missing native dependency, or a corrupted PyArrow install causes a 500.

Step 2: Set up external monitoring with Vigilmon

With /health live, point Vigilmon at it:

  1. Sign up at vigilmon.online — free tier, no credit card
  2. Click New Monitor → HTTP
  3. Enter https://yourdomain.com/health
  4. Set check interval (5 minutes on free tier)
  5. Save

Vigilmon checks from multiple geographic regions. A non-2xx response or timeout opens an incident and sends you an alert before your data warehouse starts serving stale or truncated data.

Add monitors for each critical endpoint:

| Endpoint | What it catches | |---|---| | /health | PyArrow availability, Parquet write/read round-trip | | /api/ingest | Parquet ingest service availability | | /api/query | Query service that reads Parquet partitions |

Step 3: Heartbeat monitoring for Parquet ETL jobs

Spark and Pandas ETL jobs that read and write Parquet partitions run as scheduled batch jobs with no HTTP endpoint. Heartbeat monitoring catches silent failures:

import os
import time
import io
import pyarrow as pa
import pyarrow.parquet as pq
import pyarrow.fs as pafs
import urllib.request

HEARTBEAT_URL = os.getenv("HEARTBEAT_PARQUET_ETL_URL")
S3_BUCKET = os.getenv("S3_BUCKET")
S3_PREFIX = os.getenv("S3_PREFIX", "data/events")

def validate_parquet_file(fs, path: str) -> int:
    """Read a Parquet file and return its row count. Raises on corruption."""
    with fs.open_input_file(path) as f:
        pf = pq.ParquetFile(f)
        # Read metadata to validate footer without loading all data
        meta = pf.metadata
        total_rows = meta.num_rows
        if total_rows == 0:
            raise ValueError(f"Empty file detected: {path}")
        # Read first row group to validate column integrity
        first_rg = pf.read_row_group(0)
        if first_rg.num_rows == 0:
            raise ValueError(f"First row group is empty: {path}")
        return total_rows

def run_etl():
    fs = pafs.S3FileSystem(region=os.getenv("AWS_DEFAULT_REGION", "us-east-1"))

    # Read source Parquet partitions
    source_path = f"{S3_BUCKET}/{S3_PREFIX}/raw/"
    file_info = fs.get_file_info(pafs.FileSelector(source_path, recursive=True))
    parquet_files = [f.path for f in file_info if f.path.endswith(".parquet")]

    if not parquet_files:
        raise ValueError(f"No Parquet files found at {source_path}")

    # Validate all source files before processing
    total_source_rows = 0
    for path in parquet_files:
        total_source_rows += validate_parquet_file(fs, path)

    # Process: read, transform, write
    dataset = pq.read_table(
        source_path, filesystem=fs,
        schema=pa.schema([
            pa.field("event_id", pa.string()),
            pa.field("event_ts", pa.int64()),
            pa.field("user_id", pa.string()),
            pa.field("value", pa.float64()),
        ])
    )

    # Transform
    transformed = dataset.filter(
        pa.compute.greater(dataset["value"], 0)
    )

    # Write output partitioned by date
    output_path = f"{S3_BUCKET}/{S3_PREFIX}/processed/"
    pq.write_to_dataset(
        transformed,
        root_path=output_path,
        filesystem=fs,
        compression="zstd",
    )

    # Verify output was written
    output_info = fs.get_file_info(pafs.FileSelector(output_path, recursive=True))
    output_files = [f for f in output_info if f.path.endswith(".parquet")]
    if not output_files:
        raise ValueError("ETL produced no output Parquet files")

    # Ping heartbeat only after successful verification
    if HEARTBEAT_URL:
        urllib.request.urlopen(HEARTBEAT_URL, timeout=5)

if __name__ == "__main__":
    run_etl()

In Vigilmon:

  1. Click New Monitor → Heartbeat
  2. Set the expected interval (e.g. 25 hours for a daily ETL job)
  3. Copy the unique ping URL
  4. Set the environment variable: HEARTBEAT_PARQUET_ETL_URL=https://vigilmon.online/api/heartbeat/your-unique-token

If source files are missing, a file has a corrupt footer, schema validation fails, or the output write produces no files, the heartbeat is never pinged and you get an alert after one missed interval.

Step 4: Schema validation at pipeline startup

Parquet schema mismatches are easiest to catch before your pipeline processes millions of rows:

import pyarrow as pa
import pyarrow.parquet as pq
import pyarrow.fs as pafs
import os

EXPECTED_SCHEMA = pa.schema([
    pa.field("event_id", pa.string(), nullable=False),
    pa.field("event_ts", pa.int64(), nullable=False),
    pa.field("user_id", pa.string(), nullable=True),
    pa.field("value", pa.float64(), nullable=True),
    pa.field("metadata", pa.string(), nullable=True),
])

def validate_schema_on_startup():
    fs = pafs.S3FileSystem(region=os.getenv("AWS_DEFAULT_REGION", "us-east-1"))
    source = f"{os.getenv('S3_BUCKET')}/{os.getenv('S3_PREFIX')}/raw/"

    info = fs.get_file_info(pafs.FileSelector(source, recursive=True))
    parquet_files = [f.path for f in info if f.path.endswith(".parquet")]

    if not parquet_files:
        raise RuntimeError(f"No Parquet files found at {source} — check S3 path and permissions")

    # Check schema of the first (most recent) file
    sample_path = parquet_files[-1]
    with fs.open_input_file(sample_path) as f:
        pf = pq.ParquetFile(f)
        actual_schema = pf.schema_arrow

    if not actual_schema.equals(EXPECTED_SCHEMA):
        missing = [f.name for f in EXPECTED_SCHEMA if f.name not in actual_schema.names]
        extra = [n for n in actual_schema.names if n not in EXPECTED_SCHEMA.names]
        raise RuntimeError(
            f"Parquet schema mismatch in {sample_path}. "
            f"Missing columns: {missing}. Extra columns: {extra}. "
            "Check producer for schema evolution breaking change."
        )

    print(f"Schema validated: {len(EXPECTED_SCHEMA)} columns, sample file {sample_path}")

if __name__ == "__main__":
    validate_schema_on_startup()

A schema mismatch from a producer-side breaking change now fails fast at startup rather than silently producing all-null columns for hours of processing.

Step 5: Alerts and badge embed

Slack/Discord alerts:

  1. In Vigilmon go to Notifications → New Channel
  2. Choose Slack or Discord, paste your webhook URL
  3. Enable it on your monitors

You get an instant alert when a monitor trips and a recovery notification when it's back.

Add an uptime badge to your README:

[![Uptime](https://vigilmon.online/badge/your-monitor-id.svg)](https://vigilmon.online?utm_source=devto&utm_medium=article&utm_campaign=parquet-tutorial)

What you've built

| What | How | |---|---| | External health checks | /health exercising real Parquet write/read round-trip | | Schema validation | Startup check compares actual vs expected schema | | Corrupt file detection | File validation before and after ETL processing | | ETL job monitoring | Heartbeat ping after each verified successful pipeline run | | Empty output detection | Verifies output Parquet files exist after write | | Instant alerts | Slack/Discord notifications | | README status badge | Vigilmon badge embed |

The whole setup runs on the free tier in under 30 minutes. You'll catch corrupt Parquet footers, schema evolution breaking changes, silent empty outputs, and stalled ETL pipelines before your data warehouse starts serving incorrect or stale analytics.

Get started free at vigilmon.online — monitors running in under a minute, no credit card required.

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