Approach: read Parquet, defensively detect if optional top-level fields exist, then extract/normalize using safe column expressions. Explode items into order_items; handle missing/empty arrays and null nested structs. Use Spark functions so processing is distributed and schema-drift tolerant.python
from pyspark.sql import SparkSession, functions as F, types as T
spark = SparkSession.builder.getOrCreate()
df = spark.read.parquet("s3://bucket/path/to/orders/") # or local path
cols = set(df.columns)
# helpers to safely access potentially-missing top-level columns
def col_or_literal(name, lit_value):
return F.col(name) if name in cols else F.lit(lit_value)
customer_col = col_or_literal("customer", None)
items_col = col_or_literal("items", F.array()) # empty array if missing
# normalize orders: select order-level fields and nested customer.addr fields safely
orders = df.select(
F.col("order_id"),
# customer id (if customer missing -> null)
F.when(customer_col.isNotNull(), customer_col.getField("id")).otherwise(F.lit(None)).alias("customer_id"),
# addr may be missing or null; safe getField chain with null checks
F.when(customer_col.isNotNull() & (customer_col.getField("addr").isNotNull()),
customer_col.getField("addr").getField("city")
).otherwise(F.lit(None)).alias("city"),
F.when(customer_col.isNotNull() & (customer_col.getField("addr").isNotNull()),
customer_col.getField("addr").getField("zip")
).otherwise(F.lit(None)).alias("zip"),
F.col("order_ts")
).distinct() # distinct if source has duplicates; optional
# normalize order_items: explode items (handle nulls/empty arrays)
# ensure items_col is an array: if items missing or null -> use empty array
items_safe = F.coalesce(items_col, F.array()) # if items column exists but null -> empty array
order_items = df.select(
F.col("order_id"),
F.posexplode(items_safe).alias("pos", "item") # posexplode preserves order if needed
).select(
F.col("order_id"),
# item may be null if array contained nulls
F.when(F.col("item").isNotNull(), F.col("item").getField("sku")).otherwise(F.lit(None)).alias("sku"),
F.when(F.col("item").isNotNull(), F.col("item").getField("qty")).otherwise(F.lit(None)).alias("qty")
)
# write out (example)
orders.write.mode("overwrite").parquet("s3://bucket/normalized/orders/")
order_items.write.mode("overwrite").parquet("s3://bucket/normalized/order_items/")
