Discuss error handling during PySpark data ingestion

Instruction: Explain how you would handle errors or data quality issues encountered during data ingestion in a PySpark application.

Context: The candidate is expected to outline approaches for managing errors or data quality problems when loading data into PySpark. This includes strategies for logging errors, skipping corrupt records, or cleaning data as it is ingested.

Updated

Separate transport/storage failures, parse failures and business-rule violations. Retry only transient failures with a bounded policy; permission errors, incompatible schemas and failed writes should remain visible. A successful parse does not establish key uniqueness, valid ranges or completeness.

For JSON, PERMISSIVE can retain the malformed text in a string field named by columnNameOfCorruptRecord. Include that field in an explicit schema. The following complete local fixture uses an existing classic session named spark:

from pyspark.sql import functions as F

raw = spark.sparkContext.parallelize([
    '{"id":1,"amount":3.0}',
    '{"id":2,"amount":"bad"}',
    'not-json',
    '{"id":3,"amount":null}'
])
schema = 'id long, amount double, _corrupt_record string'
parsed = spark.read.schema(schema).option('mode', 'PERMISSIVE').json(raw)
parsed = parsed.cache()
try:
    assert parsed.count() == 4
    corrupt = parsed.filter(F.col('_corrupt_record').isNotNull())
    clean_parse = parsed.filter(F.col('_corrupt_record').isNull())
    assert corrupt.count() == 2
    assert clean_parse.filter(F.col('amount').isNull()).count() == 1
finally:
    parsed.unpersist()

The explicit null is valid JSON and still needs a business policy. Persist or otherwise materialize the parsed result before queries limited to the corrupt-record column when the datasource requires it. Keep raw input and provenance for reprocessing; log safe summaries and reason counts rather than secrets or complete sensitive payloads.

FAILFAST raises on malformed records. DROPMALFORMED deliberately discards them where supported, so use it only with an accepted loss policy and reconciliation. These modes are format-specific; JSON built-in functions do not support DROPMALFORMED, and CSV has its own parsing semantics. They are not universal options for every data source.

Publish accepted data only after the quality gate passes. Reconcile input, accepted and quarantined counts, test empty/partial/retried inputs, and ensure a retry cannot publish duplicate outputs. With Spark's lazy evaluation, failures may occur at an action rather than when the reader is constructed.

JSON parser options, CSV parser options

Related Questions