Safe Schema Enforcement and Handling of Null, Malformed, and Nested Records in PySpark

Make the input contract explicit

Spark can infer a schema from JSON, but inference describes the data it inspected, not necessarily the data your application expects tomorrow. A field containing integers in one batch might contain a string in another. For a stable ingestion pipeline, define the expected types and nested structure deliberately.

Read the relevant types in A Tour of PySpark Data Types from the Apache Spark PySpark documentation. You do not need to memorize every numeric type; focus on choosing types for identifiers, amounts, and nested records.

Suppose an order feed has a customer object and an array of purchased items. We expect integer cents rather than floating-point currency values, and we include a place to retain malformed input:

from pyspark.sql import functions as F
from pyspark.sql.types import (
    ArrayType, LongType, StringType, StructField, StructType
)

item_schema = StructType([
    StructField("sku", StringType(), True),
    StructField("quantity", LongType(), True),
])

order_schema = StructType([
    StructField("event_id", StringType(), True),
    StructField("region", StringType(), True),
    StructField("amount_cents", LongType(), True),
    StructField("customer", StructType([
        StructField("id", StringType(), True),
        StructField("segment", StringType(), True),
    ]), True),
    StructField("items", ArrayType(item_schema, containsNull=True), True),
    StructField("_corrupt_record", StringType(), True),
])

customer is one nested struct; items is an array of structs. The schema describes what Spark should parse. It is not, by itself, a complete data-quality policy. In particular, declaring a field non-nullable should not replace explicit checks on externally supplied records. We will validate required fields after parsing.

Keep malformed input visible

For a typical JSON-lines feed, each line contains one complete JSON object. A different format—one JSON object spread across several lines in a file—needs Spark’s multiLine option. More importantly, you must decide what Spark should do when an input record cannot be parsed.

Watch the short comparison in Handle Corrupted Data in PySpark | Read Modes Explained by DataBeli, then use Spark’s documentation for the precise behavior with an explicit schema.

The choices have different operational meanings. PERMISSIVE lets you retain problematic input for inspection; with an explicit schema, include the configured corrupt-record string field to keep that evidence. FAILFAST stops when Spark encounters a malformed record. DROPMALFORMED discards it, which may be unacceptable when you need to account for every source row.

Here is a small, runnable example using JSON strings. One record is valid, one has a legitimate missing amount, and one has broken JSON syntax:

lines = [
    '{"event_id":"o1","region":"East","amount_cents":1300,'
    '"customer":{"id":"c1","segment":"Retail"},'
    '"items":[{"sku":"A","quantity":2}]}',

    '{"event_id":"o2","region":"East","amount_cents":null,'
    '"customer":{"id":"c2"},"items":[]}',

    '{"event_id":"o3","amount_cents":500,"items":[',
]

raw = spark.createDataFrame([(line,) for line in lines], ["raw_json"])

parsed = (
    raw.withColumn(
        "order",
        F.from_json(
            F.col("raw_json"),
            order_schema,
            {
                "mode": "PERMISSIVE",
                "columnNameOfCorruptRecord": "_corrupt_record",
            },
        ),
    )
    .select("raw_json", "order.*")
)

parsed.select(
    "event_id", "amount_cents", "_corrupt_record"
).show(truncate=False)

The key distinction is that o2 has valid JSON with a null amount, while the broken o3 line is a parse problem. Both may have null parsed fields, so testing amount_cents.isNull() alone cannot explain what went wrong. Keep the original string and the corrupt-record field until you have classified the row.

For JSON files rather than an existing string column, the corresponding read uses the same schema and options:

file_orders = (
    spark.read
    .schema(order_schema)
    .option("mode", "PERMISSIVE")
    .option("columnNameOfCorruptRecord", "_corrupt_record")
    .json("path/to/order-feed/*.json")
)

Replace the illustrative path with an actual accessible input path. Retaining source files or a raw landing copy matters: a corrupt-record field preserves malformed JSON, but it does not automatically preserve the original text of every well-formed row that later fails a business rule.

Separate parse failures from business-rule failures

A successful parse means the record fits the format and types well enough to read; it does not mean the record is usable. For this example, suppose an order requires a nonblank event_id, a nonnegative amount_cents, and a customer ID. A missing region or customer segment may be acceptable for reporting, but those rules would need confirmation from the business owner.

Give each rejected row a reason before producing the trusted output:

classified = parsed.withColumn(
    "reject_reason",
    F.when(
        F.col("_corrupt_record").isNotNull(),
        F.lit("parse_error"),
    )
    .when(
        F.col("event_id").isNull()
        | (F.trim(F.col("event_id")) == ""),
        F.lit("missing_event_id"),
    )
    .when(
        F.col("amount_cents").isNull()
        | (F.col("amount_cents") < 0),
        F.lit("invalid_amount"),
    )
    .when(
        F.col("customer.id").isNull(),
        F.lit("missing_customer_id"),
    )
    .otherwise(F.lit(None).cast("string"))
)

rejected = classified.filter(F.col("reject_reason").isNotNull()).select(
    "raw_json", "_corrupt_record", "reject_reason"
)

accepted = classified.filter(F.col("reject_reason").isNull())

The order of when clauses is intentional: a parse error takes precedence over downstream checks on fields that may be null because parsing failed. Here, o2 is rejected for its missing amount; the broken o3 input is classified as a parse error. In a production pipeline, preserve rejected in a controlled quarantine dataset with source and ingestion metadata, and measure its count. Do not silently drop it or mix it into a trusted table.

Nulls require a second distinction: unknown is not zero. Replacing a missing amount with zero would invent a financial value. For an optional reporting label, a documented default can be reasonable:

clean_orders = accepted.select(
    "event_id",
    F.coalesce(F.col("region"), F.lit("Unknown")).alias("region"),
    "amount_cents",
    F.col("customer.id").alias("customer_id"),
    F.coalesce(
        F.col("customer.segment"), F.lit("Unspecified")
    ).alias("customer_segment"),
    "items",
)

Apply defaults only where they have a defined meaning. Use isNull() and isNotNull() for null checks; ordinary equality comparisons do not classify nulls the way comparisons with non-null values do.

Sometimes a source contract supplies amounts as strings, so you must convert them before applying the amount rule. A cast can behave differently depending on Spark’s ANSI setting: an invalid value may raise an error or become null. Read the caution in the PySpark types guide, then consider an explicit safe conversion.

amount_text = spark.createDataFrame(
    [("1200",), ("oops",), (None,)],
    ["amount_text"],
)

checked_amounts = amount_text.withColumn(
    "amount_cents",
    F.expr("try_cast(amount_text as bigint)"),
).withColumn(
    "conversion_failed",
    F.col("amount_text").isNotNull()
    & F.col("amount_cents").isNull(),
)

try_cast makes the conversion failure inspectable as a null result, while conversion_failed distinguishes "oops" from an originally missing value. The appropriate response—reject, quarantine, or correct upstream—still depends on the data contract.

Extract nested data without changing the grain accidentally

Once accepted orders are available, nested fields are ordinary typed columns. customer.id selects a field from a struct. An array requires a further decision: do you want one row per order or one row per item?

item_rows = (
    clean_orders
    .select(
        "event_id",
        F.explode("items").alias("item"),
    )
    .select(
        "event_id",
        F.col("item.sku").alias("sku"),
        F.col("item.quantity").alias("quantity"),
    )
)

explode produces a row for each array element. An order with two items becomes two rows; an order with an empty or null array produces no item rows. That is appropriate for an item-level table, but not if you need to retain every order in the resulting dataset. explode_outer can retain a parent row when the array is null or empty, with null item fields. State the intended grain before choosing between them.

A typed array does not guarantee that every item satisfies a business rule. If item rows will be served as trusted data, decide whether a missing SKU, null item, or nonpositive quantity should reject the whole order or only the affected item. The schema, the order-level checks, and the item-level checks each answer a different question.

Conclusion

A robust PySpark ingestion path has three layers: an explicit type and nesting contract, a deliberate malformed-record policy, and business validation before data becomes trusted. Preserve enough raw evidence to tell a parse failure from a genuinely missing value; do not substitute convenient defaults for required facts. When flattening nested arrays, name the output grain so row-count changes are expected rather than surprising.

You have now covered the DataFrame foundations in this module. Next, the course turns to performance: how partition choices affect the work Spark performs on these validated datasets.