Handling bad records: a practical Spark & Data Eng guide

Introduction

A production Spark job runs smoothly for weeks, then one malformed record crashes your entire pipeline at 3 AM. Handling bad records: a practical Spark & Data Eng guide equips you with the strategies and configurations to prevent that scenario. In real-world data engineering, you won't eliminate bad data—you'll architect around it.

Whether you're ingesting third-party APIs, processing user uploads, or consuming Kafka topics, corrupt records are inevitable. The question isn't whether they'll arrive, but how your system responds. We'll walk through proven patterns that let one row fail gracefully without breaking the job, keeping your pipelines resilient and observable.

PERMISSIVE mode with corrupt record columns

By default, Spark's JSON and CSV readers use PERMISSIVE mode, which attempts to parse each record. When a record is malformed, instead of throwing an exception, Spark stashes the raw input in a special column—typically _corrupt_record. This is your first line of defense.

Here's what happens in practice. You're reading a CSV where one row has mismatched quote escaping:

from pyspark.sql import SparkSession
from pyspark.sql.types import StructType, StructField, StringType, IntegerType

spark = SparkSession.builder.appName("corrupt_records").getOrCreate()

# Sample data with one malformed row
data = """id,name,email
1,Alice,alice@example.com
2,Bob "The Builder" Smith,bob@example.com
3,Charlie,charlie@example.com"""

with open("/tmp/data.csv", "w") as f:
    f.write(data)

# Read with PERMISSIVE mode (default)
df = spark.read \
    .option("header", "true") \
    .option("mode", "PERMISSIVE") \
    .option("columnNameOfCorruptRecord", "_corrupt_record") \
    .csv("/tmp/data.csv")

df.show(truncate=False)
# Output shows id, name, email columns plus _corrupt_record
# Row with mismatched quotes will have nulls in structured columns
# and the full raw line in _corrupt_record

The benefit is clear: your job completes. You get partial data from rows that parse correctly, and the raw malformed data preserved for inspection. The downside is you must actively handle these records—they won't magically transform into valid rows.

In production, filter these corrupt records into a separate dataframe for logging and later analysis:

corrupt_rows = df.filter(df._corrupt_record.isNotNull())
valid_rows = df.filter(df._corrupt_record.isNull()).drop("_corrupt_record")

print(f"Valid records: {valid_rows.count()}")
print(f"Corrupt records: {corrupt_rows.count()}")

# Log corrupt records for debugging
corrupt_rows.coalesce(1).write \
    .mode("overwrite") \
    .option("header", "true") \
    .csv("/tmp/corrupt_records")

This separation keeps your downstream processing clean while preserving evidence. However, PERMISSIVE mode has limits—it won't rescue malformed Parquet or ORC files, and it catches structural problems but not semantic ones (like invalid dates or out-of-range integers).

badRecordsPath on some platforms

Some managed Spark platforms (notably Databricks, but increasingly others) offer a native badRecordsPath configuration that automatically redirects corrupt records to a separate location on disk. This is different from the in-column approach—the bad records are written out-of-band as you read.

Here's how it works on Databricks:

df = spark.read \
    .option("header", "true") \
    .option("badRecordsPath", "/mnt/bad_records/daily_import") \
    .option("mode", "PERMISSIVE") \
    .json("/mnt/data/daily_import.json")

# Corrupt records are automatically written to /mnt/bad_records/daily_import
# They appear as JSON files with metadata about why they failed
# Your dataframe contains only successfully parsed rows

The advantage is automation—you don't manually separate corrupt rows. The platform handles it, and you can inspect the bad records location after the job runs. You also get structured error metadata explaining what went wrong.

However, this feature isn't universally available. Open-source Spark running on HDFS or S3 doesn't support it natively. If you're on a custom Kubernetes cluster or legacy Spark deployment, you'll need the manual _corrupt_record approach. Always check your platform's documentation before relying on badRecordsPath.

A practical pattern: use it where available, but code your pipelines to work without it. This keeps you portable and defensive.

Quarantine tables

For serious production systems, storing bad records in ephemeral locations (temporary CSVs, logs) isn't enough. You need a persistent quarantine—a structured table where bad records accumulate, can be inspected, and are tied to the ingestion batch.

Design a quarantine table like this:

from pyspark.sql.functions import current_timestamp, lit
from datetime import datetime

# Create quarantine table structure
quarantine_schema = """
    batch_id STRING,
    job_name STRING,
    error_reason STRING,
    raw_record STRING,
    attempt_count INT,
    first_seen TIMESTAMP,
    last_seen TIMESTAMP,
    resolved BOOLEAN
"""

spark.sql(f"""
    CREATE TABLE IF NOT EXISTS bad_records_quarantine (
        {quarantine_schema}
    )
    USING DELTA
    PARTITIONED BY (batch_id)
""")

# When reading data, collect corrupt records
batch_id = datetime.now().strftime("%Y%m%d_%H%M%S")

corrupt_df = df.filter(df._corrupt_record.isNotNull()) \
    .select(
        lit(batch_id).alias("batch_id"),
        lit("daily_import_job").alias("job_name"),
        lit("parse_error").alias("error_reason"),
        df._corrupt_record.alias("raw_record"),
        lit(1).alias("attempt_count"),
        current_timestamp().alias("first_seen"),
        current_timestamp().alias("last_seen"),
        lit(False).alias("resolved")
    )

# Insert into quarantine
corrupt_df.write.mode("append").insertInto("bad_records_quarantine")

print(f"Quarantined {corrupt_df.count()} records in batch {batch_id}")

The quarantine table gives you several advantages. First, it's queryable—you can run analytics on which records fail and why. Second, it's persistent across job runs, so you can track whether the same record fails repeatedly (suggesting a systematic issue). Third, you can add a resolved flag and a support process: engineers can investigate, fix upstream data, then mark records as resolved.

In practice, many teams integrate quarantine tables with their data quality dashboards. They set up automatic notifications when quarantine tables grow beyond expected thresholds, triggering investigation before problems cascade.

Alert when bad record rates rise

The final piece is observability. A few corrupt records every day might be normal. But if your corrupt record rate jumps from 0.1% to 5%, something broke upstream—and you need to know immediately.

Implement a simple metric check after each ingestion:

from pyspark.sql.functions import count as spark_count

total_records = df.count()
corrupt_count = df.filter(df._corrupt_record.isNotNull()).count()
valid_count = total_records - corrupt_count

corrupt_rate = corrupt_count / total_records if total_records > 0 else 0

print(f"Total: {total_records}, Valid: {valid_count}, Corrupt: {corrupt_count}")
print(f"Corrupt rate: {corrupt_rate:.2%}")

# Define thresholds
warning_threshold = 0.01  # 1%
error_threshold = 0.05    # 5%

if corrupt_rate > error_threshold:
    # Send critical alert
    send_alert(
        severity="CRITICAL",
        message=f"Bad record rate {corrupt_rate:.2%} exceeds threshold {error_threshold:.2%}",
        batch_id=batch_id,
        corrupt_count=corrupt_count
    )
elif corrupt_rate > warning_threshold:
    send_alert(
        severity="WARNING",
        message=f"Bad record rate {corrupt_rate:.2%} elevated",
        batch_id=batch_id
    )

Store these metrics in a monitoring system (CloudWatch, Datadog, Prometheus, or custom logging). Set up dashboards that visualize corrupt rate over time, and alert rules that trigger notifications to your team's Slack channel or on-call rotation.

A concrete example: one team discovered that a third-party API started sending microseconds in timestamp fields instead of milliseconds. Their corrupt rate jumped from 0.02% to 12% in a single day. Because they'd instrumented alerting, they caught it within 30 minutes and contacted the API vendor for a fix. Without alerting, the bad data would have silently accumulated for days, potentially affecting downstream analytics and reporting.

For best results, track corrupt rates per source, per data type, and per job. Different sources have different acceptable baselines—user-uploaded CSVs might have 2-3% corruption naturally, while database exports should be nearly 0%.

Common mistakes

• Ignoring corrupt records silently: Using DROPMALFORMED mode feels safe because there's no error, but you lose data without knowing. Always use PERMISSIVE or similar and explicitly handle the bad records.
• Not logging enough context: If you only store the raw record without the batch ID, job name, or timestamp, debugging later becomes a nightmare. Always attach metadata.
• Treating all corruption equally: A malformed JSON is different from an out-of-range integer. Distinguish error types so you can route them to the right team or automation.
• Setting thresholds too low or too high: A 0.001% threshold will trigger false alarms and fatigue your team. A 10% threshold will miss real problems. Calibrate based on your data source's historical baseline.
• Skipping downstream validation: Even valid-looking records can contain semantic errors (negative ages, future dates, malformed IDs). Bad record handling at the source is necessary but not sufficient.

Key takeaways

• Use PERMISSIVE mode with a _corrupt_record column (or badRecordsPath on supported platforms) to let your job complete even when records fail to parse.
• Actively separate valid and corrupt records, then route corrupt records to a quarantine table or dedicated storage for investigation.
• Design your quarantine to include batch ID, error reason, timestamp, and resolution status—make it queryable and actionable.
• Measure corrupt record rates per job and source, then alert when rates exceed historical baselines. Automation catches problems early.
• Remember that handling corrupt records is one layer of a multi-layered data quality strategy. Combine it with schema validation, semantic checks, and downstream monitoring.