Content
Table of Contents
1. Background
1.1 Overview
When ingesting event data (JSON) into the bronze layer, “did the load succeed?” is not a yes/no question. A run can finish without throwing an exception and still be silently wrong: files can go missing in transit, and Spark’s from_json will quietly drop any field that is not in the schema you hand it. Neither shows up as an error — the job goes green while data is lost.
So we made quality checks metadata-driven and first-class: the source system ships metadata alongside the data, we compare the actual data against that metadata before trusting it, and we record the outcome of every object into a run_control table. That table becomes the single source of truth for “what loaded, how much, and why it failed”.
There are two independent checks, run at two different stages:
- File completeness — did every file the source says it sent actually arrive, intact? (count + bytes)
- Schema conformance — does every field present in the JSON payload exist in our declared event schema? (no silent field drops)
Every object’s result — successful, failed, or skipped — plus row counts and an error message, is written to run_control.
1.2 Why metadata-driven
The source system writes, next to each batch of data files, a set of metadata sidecar files listing every file it produced and each file’s expected size in bytes. That metadata is the contract. Instead of guessing whether a batch is complete, we validate the delivered files against the contract, and we validate the delivered fields against our own schema definitions. Nothing is assumed — everything is compared.
2. The quality problem
Two failure modes are invisible without explicit checks:
-
Missing or truncated files. A batch of thousands of files can lose a few in transit, or a file can be partially written. Spark will happily read whatever is there and report success on a partial dataset.
-
Silent field drops.
from_json(col, schema)only keeps fields that are declared inschema. If the source adds a new field (schema drift) and we haven’t updated our event schema, that field is dropped without warning. The row count looks fine; the data is incomplete.
Both must be turned into loud, recorded failures instead of silent data loss.
3. Technical solution
3.1 Check 1 — file completeness against metadata (count + bytes)
Each batch ships metadata sidecar files that list every delivered file and its sizeBytes. We read that metadata, list the files actually present in storage, join the two, and flag anything missing or with a size mismatch. If either exists, we fail the run before any data is written.
# expected files (from metadata sidecars): destination path + expected size in bytes
metadata_df = (
spark.read.option("multiline", "true").json(metadata_folder_path)
.withColumn("metadata_file_path", F.col("_metadata.file_path"))
.select(F.col("metadata_file_path"), F.explode("files").alias("file"))
.select(
F.col("metadata_file_path"),
F.col("file.destinationPath").alias("destination_path"),
F.col("file.sizeBytes").cast("long").alias("expected_size"),
)
.withColumn("relative_path", F.regexp_extract("destination_path", r".*/Raw/(.*)", 1))
.withColumn("full_path", F.concat(F.lit(raw_folder_path), F.col("relative_path")))
)
# actual files present in storage + their real size on disk
actual_files_df = (
spark.read.format("binaryFile").option("recursiveFileLookup", "true").load(raw_folder_path)
.select(F.col("path").alias("full_path"), F.col("length").cast("long").alias("actual_size"))
)
# compare expected vs actual
comparison_df = metadata_df.join(actual_files_df, "full_path", "left")
missing_files = comparison_df.filter(F.col("actual_size").isNull())
size_mismatches = comparison_df.filter(
F.col("actual_size").isNotNull() & (F.col("expected_size") != F.col("actual_size"))
)
if missing_files.count() > 0 or size_mismatches.count() > 0:
errors = [f"File not found: {r.full_path} (metadata {r.metadata_file_path})"
for r in missing_files.collect()]
errors += [f"Size mismatch for {r.full_path}: expected {r.expected_size}, got {r.actual_size}"
for r in size_mismatches.collect()]
raise ValueError("File completeness check failed:\n" + "\n".join(errors))
Why bytes, not just count? A file can exist but be half-written (truncated upload). Comparing sizeBytes catches partial files that a pure count check would miss. This check is a hard gate: if it fails, we stop — there is no point loading a known-incomplete batch.
3.2 Check 2 — schema conformance (no silent field drops)
The subtle one. from_json silently drops fields not in the schema, so before parsing we compare the fields actually present in the JSON payload against the fields declared in our event schema. Any field in the data but not in the schema is an unknown_field — a signal of source-side schema drift that would otherwise be lost.
We infer the real payload schema across all rows (serverless-safe: no RDD, no spark.read.json) using schema_of_variant_agg, then flatten both the inferred schema and the declared schema to dotted field paths and take the set difference.
def flatten_field_paths(data_type, prefix=""):
# collect all (nested) field paths of a Struct/Array as a set of dotted names
if isinstance(data_type, ArrayType):
data_type = data_type.elementType
paths = set()
if isinstance(data_type, StructType):
for field in data_type.fields:
path = f"{prefix}{field.name}"
paths.add(path)
paths |= flatten_field_paths(field.dataType, f"{path}.")
return paths
def get_payload_field_paths(df):
# infer the actual payload schema merged across ALL rows (serverless-safe)
inferred_ddl = df.select(F.expr("schema_of_variant_agg(parse_json(Payload))").alias("s")).first()["s"]
if inferred_ddl is None: # no rows / all-null payloads
return set()
inferred_ddl = inferred_ddl.replace("OBJECT<", "STRUCT<") # make it castable
inferred_type = spark.sql(f"SELECT CAST(NULL AS {inferred_ddl}) AS s").schema.fields[0].dataType
return flatten_field_paths(inferred_type)
# the actual check: fields in the JSON but NOT in our event schema
unknown_fields = get_payload_field_paths(df) - flatten_field_paths(schema)
if unknown_fields:
# record as FAILED with the offending field list -> nothing silently lost
record_failure(object_name, f"Fields in json but not in event schema: {sorted(unknown_fields)}")
else:
record = process_one_object(object_name, df, schema)
This deliberately checks one direction (data ⊄ schema). Fields that are in the schema but absent from the data are fine — they just come through as null. What we cannot tolerate is a field arriving that we would silently throw away.
3.3 Recording the outcome in run_control
Every object produces a record regardless of outcome, and each record maps to one row in the run_control table. Three terminal states:
| Status | Meaning |
|---|---|
successful |
Passed both checks and was written to bronze. |
failed |
A check failed (unknown fields) or the write raised. error_message explains why. |
skipped |
No event schema exists for this topic; nothing was written. |
We also capture input_row_count / insert_row_count, timing (load_start, load_end, duration_seconds), and a last_successful_run_flag so downstream consumers can always find the latest good load per object.
def process_one_object(object_name, df, schema):
load_start = datetime.now(AMSTERDAM_TZ)
record = {"run_control_id": str(uuid.uuid4()), "object_name": object_name, "load_start": load_start}
try:
root_count = process_topic(df=df, schema=schema, topic=object_name)
record.update(status="successful",
input_row_count=root_count, insert_row_count=root_count, ...)
except Exception as e:
record.update(status="failed", error_message=str(e), ...)
return record
The rows are collected in memory during processing and written to run_control once, sequentially, afterwards. Before marking a new success as the latest, we demote the previous one:
for rec in run_control_records:
row = penu.RunControlModel(
run_control_id=rec["run_control_id"], source_name=source_name,
input_object=rec["object_name"], output_object=rec["object_name"],
output_layer=target_layer, status=rec["status"],
load_start=rec["load_start"], load_end=rec["load_end"],
duration_seconds=rec["duration_seconds"],
last_successful_run_flag=(rec["status"] == "successful"),
error_message=rec.get("error_message"),
input_row_count=rec.get("input_row_count"),
insert_row_count=rec.get("insert_row_count"),
...
)
if row.last_successful_run_flag:
# only one "latest success" per (source, object, layer)
run_control_agent.reset_last_successful_run_flag_false(source_name, rec["object_name"], target_layer)
run_control_agent.insert_row(row)
4. The full flow
flowchart TD
A[Batch arrives: data files + metadata sidecars] --> B[Check 1: file completeness]
B -->|missing file or size mismatch| B1[raise ValueError - hard stop]
B -->|all files present and intact| C[Read + flatten payloads per topic]
C --> D{Event schema exists for topic?}
D -->|no| S[status = skipped]
D -->|yes| E[Check 2: payload fields vs event schema]
E -->|unknown fields found| F[status = failed + error_message]
E -->|conformant| G[Write to bronze Delta table]
G -->|write error| F
G -->|ok| H[status = successful + row counts]
S --> R[(run_control table)]
F --> R
H --> R
R --> Z[Latest good load per object via last_successful_run_flag]
And how a single object moves through the two gates:
sequenceDiagram
participant M as Metadata sidecar
participant S as Storage
participant J as Job
participant RC as run_control
M->>J: expected files + sizeBytes
S->>J: actual files + length
J->>J: Check 1 - completeness (count + bytes)
Note over J: fail -> hard stop for the whole batch
J->>J: infer payload fields (schema_of_variant_agg)
J->>J: Check 2 - payload fields - event schema
alt unknown fields
J->>RC: status = failed (+ offending fields)
else conformant
J->>J: write to bronze
J->>RC: status = successful (+ row counts)
end
5. Design decisions worth calling out
- Two gates, two blast radii. File completeness is a batch-level hard stop (a
ValueErrorthat aborts the whole run) because a missing file poisons everything. Schema conformance is per-object: one drifting topic is recorded asfailedwhile the other topics still load, so one bad topic doesn’t block the rest. - Serverless-safe inference.
schema_of_variant_agg(parse_json(...))merges the payload schema across all rows without RDDs orspark.read.json, both of which are constrained on serverless. TheOBJECT<...>→STRUCT<...>rewrite makes the inferred DDL castable so we can turn it back into a realDataType. - Everything is recorded, nothing is guessed. Even a
skippedtopic (no schema) gets arun_controlrow, so an operator can see exactly which topics were ignored and why. last_successful_run_flaggives downstream consumers a reliable pointer to the newest good load per object without scanning history.
6. Test focus
- Completeness gate: delete one delivered file and truncate another; confirm both are reported and the run hard-stops with a clear message.
- Schema drift: add a field to the JSON payload that is not in the event schema; confirm the object is
failedwith the exact unknown-field list, and other topics still succeed. - Skipped path: send a topic with no matching event schema; confirm a
skippedrow with a reason. - run_control integrity: verify counts, timings,
error_message, and that exactly one row per object carrieslast_successful_run_flag = trueafter a successful reload. - Empty / all-null payloads: confirm
get_payload_field_pathsreturns an empty set and does not false-positive.
7. Other issues
- Should schema drift always fail, or should additive-only drift (new nullable field) be allowed and just logged? Currently any unknown field fails — pending discussion with the source supplier.
- Consider promoting the field-path comparison into a shared, recursive
get_schema_diffutility so raw, bronze, and metadata checks all use one implementation. - Surface
run_controlfailures to alerting so afailed/skippedobject pages someone instead of waiting for a consumer to notice.