intermediateView Python source
Normalize nested order events without losing rejected records
Read JSONL files, validate complete orders, then use Spark SQL and DataFrame code on Sail to separate order headers from line items. Every rejected occurrence keeps its original bytes, source filename, source SHA-256 and physical line number. This makes an invalid child's effect on its whole order visible.
Run it
Follow the repository setup, then run all commands below from the repository root (sail-examples/):
uv run pipelines/normalize-order-events/main.py
The input is an authored 15-record JSONL snapshot.
The program copies it into temporary input files, reads those actual files
inside mapInArrow, writes typed Parquet outputs and reads them back. The fixture
needs no data download, key or GPU. uv installs the pinned packages on first run.
The default run prints the orders, lines and rejects tables: three clean
orders, three clean line items and twelve rejected source occurrences.
For a compact count check:
uv run pipelines/normalize-order-events/main.py --json
Expected output:
{"lines": 3, "orders": 3, "rejects": 12}
For a larger local check, create independent copies with different order IDs:
uv run pipelines/normalize-order-events/main.py --copies 100 --json
This processes 1,500 physical records and returns counts of 300 orders, 300 line items and 1,200 rejects. The limit is 1,000 copies. This exercises repeated file ingestion and normalization, not production throughput or arbitrary event data.
Save your JSONL files as pipelines/normalize-order-events/orders.jsonl and
pipelines/normalize-order-events/more-orders.jsonl, then run:
uv run pipelines/normalize-order-events/main.py \
--input pipelines/normalize-order-events/orders.jsonl \
--input pipelines/normalize-order-events/more-orders.jsonl \
--output-dir pipelines/normalize-order-events/.data/normalized-orders \
--json
These files must follow the supported contract below. --input cannot be
combined with --copies. --output-dir retains the orders/, lines/, and rejects/ Parquet datasets
under pipelines/normalize-order-events/.data/normalized-orders/ after process exit.
That output directory must not already exist. Writes are not an atomic publication protocol: failure can leave partial
output, so inspect it and choose a new directory when retrying. Without this
option, output datasets are temporary and removed at process exit.
Output
The newest capture appears first. Older output stays here when a result changes.
Current output · Sail 0.7.3
Captured on Sail 0.7.3. Latest capture: 2026-11-05, Python 3.12.14.
orders
+----------------+----------------------------------------------------------------+-----------+------------+-----------+--------+----------+-----------+--------------------------------------------------------------------------------+--------------------------------------------------------------------------------+
| source_id| source_sha256|line_number| order_id|customer_id|currency|item_count|gross_minor| raw_record_base64| raw_json|
+----------------+----------------------------------------------------------------+-----------+------------+-----------+--------+----------+-----------+--------------------------------------------------------------------------------+--------------------------------------------------------------------------------+
|orders-000.jsonl|baffe6dcc7fc6ae30fa2c561e323c1daa1de9180f5c4b1a13a5277a4ce166783| 1|ORDER-A-0000| CUSTOMER-1| USD| 2| 3000|eyJzY2hlbWFfdmVyc2lvbiI6MSwib3JkZXJfaWQiOiJPUkRFUi1BLTAwMDAiLCJjdXN0b21lcl9pZ...|{"schema_version":1,"order_id":"ORDER-A-0000","customer_id":"CUSTOMER-1","cur...|
|orders-000.jsonl|baffe6dcc7fc6ae30fa2c561e323c1daa1de9180f5c4b1a13a5277a4ce166783| 2|ORDER-B-0000| CUSTOMER-2| GBP| 0| 0|eyJzY2hlbWFfdmVyc2lvbiI6MSwib3JkZXJfaWQiOiJPUkRFUi1CLTAwMDAiLCJjdXN0b21lcl9pZ...|{"schema_version":1,"order_id":"ORDER-B-0000","customer_id":"CUSTOMER-2","cur...|
|orders-000.jsonl|baffe6dcc7fc6ae30fa2c561e323c1daa1de9180f5c4b1a13a5277a4ce166783| 15|ORDER-Z-0000| CUSTOMER-1| USD| 1| 0|eyJzY2hlbWFfdmVyc2lvbiI6MSwib3JkZXJfaWQiOiJPUkRFUi1aLTAwMDAiLCJjdXN0b21lcl9pZ...|{"schema_version":1,"order_id":"ORDER-Z-0000","customer_id":"CUSTOMER-1","cur...|
+----------------+----------------------------------------------------------------+-----------+------------+-----------+--------+----------+-----------+--------------------------------------------------------------------------------+--------------------------------------------------------------------------------+
lines
+----------------+----------------------------------------------------------------+-----------+------------+-------+------+--------+----------------+-----------+
| source_id| source_sha256|line_number| order_id|line_id| sku|quantity|unit_price_minor|gross_minor|
+----------------+----------------------------------------------------------------+-----------+------------+-------+------+--------+----------------+-----------+
|orders-000.jsonl|baffe6dcc7fc6ae30fa2c561e323c1daa1de9180f5c4b1a13a5277a4ce166783| 1|ORDER-A-0000| L1|WIDGET| 2| 1250| 2500|
|orders-000.jsonl|baffe6dcc7fc6ae30fa2c561e323c1daa1de9180f5c4b1a13a5277a4ce166783| 1|ORDER-A-0000| L2| BAG| 1| 500| 500|
|orders-000.jsonl|baffe6dcc7fc6ae30fa2c561e323c1daa1de9180f5c4b1a13a5277a4ce166783| 15|ORDER-Z-0000| FREE|SAMPLE| 1| 0| 0|
+----------------+----------------------------------------------------------------+-----------+------------+-------+------+--------+----------------+-----------+
rejects
+----------------+----------------------------------------------------------------+-----------+--------------------------------------------------------------------------------+--------------------------------------------------------------------------------+-----------------------------------+
| source_id| source_sha256|line_number| raw_record_base64| raw_json| reasons|
+----------------+----------------------------------------------------------------+-----------+--------------------------------------------------------------------------------+--------------------------------------------------------------------------------+-----------------------------------+
|orders-000.jsonl|baffe6dcc7fc6ae30fa2c561e323c1daa1de9180f5c4b1a13a5277a4ce166783| 3|eyJzY2hlbWFfdmVyc2lvbiI6MSwib3JkZXJfaWQiOiJPUkRFUi1DLTAwMDAiLCJjdXN0b21lcl9pZ...|{"schema_version":1,"order_id":"ORDER-C-0000","customer_id":"CUSTOMER-1","cur...| [items[0].invalid_quantity]|
|orders-000.jsonl|baffe6dcc7fc6ae30fa2c561e323c1daa1de9180f5c4b1a13a5277a4ce166783| 4|eyJzY2hlbWFfdmVyc2lvbiI6MSwib3JkZXJfaWQiOiJPUkRFUi1ELTAwMDAiLCJjdXN0b21lcl9pZ...|{"schema_version":1,"order_id":"ORDER-D-0000","customer_id":"CUSTOMER-1","cur...| [items[1].duplicate_line_id]|
|orders-000.jsonl|baffe6dcc7fc6ae30fa2c561e323c1daa1de9180f5c4b1a13a5277a4ce166783| 5|eyJzY2hlbWFfdmVyc2lvbiI6MSwib3JkZXJfaWQiOiJPUkRFUi1FLTAwMDAiLCJjdXN0b21lcl9pZ...|{"schema_version":1,"order_id":"ORDER-E-0000","customer_id":"CUSTOMER-1","cur...| [missing_items]|
|orders-000.jsonl|baffe6dcc7fc6ae30fa2c561e323c1daa1de9180f5c4b1a13a5277a4ce166783| 6|eyJzY2hlbWFfdmVyc2lvbiI6MSwib3JkZXJfaWQiOiJPUkRFUi1GLTAwMDAiLCJjdXN0b21lcl9pZ...|{"schema_version":1,"order_id":"ORDER-F-0000","customer_id":"CUSTOMER-1","cur...| [null_items]|
|orders-000.jsonl|baffe6dcc7fc6ae30fa2c561e323c1daa1de9180f5c4b1a13a5277a4ce166783| 7|eyJzY2hlbWFfdmVyc2lvbiI6Miwib3JkZXJfaWQiOiJPUkRFUi1HLTAwMDAiLCJjdXN0b21lcl9pZ...|{"schema_version":2,"order_id":"ORDER-G-0000","customer_id":"CUSTOMER-1","cur...| [unsupported_schema_version]|
|orders-000.jsonl|baffe6dcc7fc6ae30fa2c561e323c1daa1de9180f5c4b1a13a5277a4ce166783| 8| eyJzY2hlbWFfdmVyc2lvbiI6MSwib3JkZXJfaWQiOiJPUkRFUi1ILTAwMDAiLAo=| {"schema_version":1,"order_id":"ORDER-H-0000",\n| [malformed_json]|
|orders-000.jsonl|baffe6dcc7fc6ae30fa2c561e323c1daa1de9180f5c4b1a13a5277a4ce166783| 9|eyJzY2hlbWFfdmVyc2lvbiI6MSwib3JkZXJfaWQiOiJPUkRFUi1JLTAwMDAiLCJvcmRlcl9pZCI6I...|{"schema_version":1,"order_id":"ORDER-I-0000","order_id":"OTHER-0000","custom...| [duplicate_json_member]|
|orders-000.jsonl|baffe6dcc7fc6ae30fa2c561e323c1daa1de9180f5c4b1a13a5277a4ce166783| 10|eyJzY2hlbWFfdmVyc2lvbiI6MSwib3JkZXJfaWQiOiJPUkRFUi1KLTAwMDAiLCJjdXN0b21lcl9pZ...|{"schema_version":1,"order_id":"ORDER-J-0000","customer_id":"CUSTOMER-1","cur...|[items[0].invalid_unit_price_minor]|
|orders-000.jsonl|baffe6dcc7fc6ae30fa2c561e323c1daa1de9180f5c4b1a13a5277a4ce166783| 11|eyJzY2hlbWFfdmVyc2lvbiI6MSwib3JkZXJfaWQiOiJPUkRFUi1YLTAwMDAiLCJjdXN0b21lcl9pZ...|{"schema_version":1,"order_id":"ORDER-X-0000","customer_id":"CUSTOMER-1","cur...| [duplicate_order_id]|
|orders-000.jsonl|baffe6dcc7fc6ae30fa2c561e323c1daa1de9180f5c4b1a13a5277a4ce166783| 12|eyJzY2hlbWFfdmVyc2lvbiI6MSwib3JkZXJfaWQiOiJPUkRFUi1YLTAwMDAiLCJjdXN0b21lcl9pZ...|{"schema_version":1,"order_id":"ORDER-X-0000","customer_id":"CUSTOMER-1","cur...| [duplicate_order_id]|
|orders-000.jsonl|baffe6dcc7fc6ae30fa2c561e323c1daa1de9180f5c4b1a13a5277a4ce166783| 13| W10K| []\n| [top_level_not_object]|
|orders-000.jsonl|baffe6dcc7fc6ae30fa2c561e323c1daa1de9180f5c4b1a13a5277a4ce166783| 14| Cg==| \n| [empty_record]|
+----------------+----------------------------------------------------------------+-----------+--------------------------------------------------------------------------------+--------------------------------------------------------------------------------+-----------------------------------+
Python code
main.py405 lines
# /// script
# requires-python = ">=3.12"
# dependencies = [
# "pysail==0.7.3",
# "pyspark-client==4.1.3",
# "pyarrow==26.0.0",
# "pandas==3.0.6",
# ]
# ///
"""Normalize a bounded JSONL snapshot into orders, lines and original rejected records."""
import argparse
import atexit
import base64
import hashlib
import json
import re
import shutil
import tempfile
from collections import Counter
from decimal import Decimal, InvalidOperation
from pathlib import Path
MAX_FILE_BYTES = 1_048_576
MAX_RECORD_BYTES = 65_536
MAX_RECORDS_PER_FILE = 10_000
MAX_INTEGER_DIGITS = 4300
MAX_ITEMS = 100
MAX_QUANTITY = 1_000_000
MAX_PRICE = 1_000_000_000
MAX_FILES = 100
ID = re.compile(r"[A-Za-z0-9][A-Za-z0-9_-]{0,63}\Z")
SOURCE = "source_id string, source_sha256 string, line_number long"
EVIDENCE = "raw_record_base64 string, raw_json string"
ITEM = "struct<line_id:string,sku:string,quantity:bigint,unit_price_minor:bigint>"
PARSED_SCHEMA = (
SOURCE
+ ", "
+ EVIDENCE
+ ", order_id string, customer_id string, currency string, "
+ "items array<"
+ ITEM
+ ">, reasons array<string>"
)
ORDER_SCHEMA = (
SOURCE
+ ", order_id string, customer_id string, currency string, item_count long, gross_minor long, "
+ EVIDENCE
)
LINE_SCHEMA = (
SOURCE
+ ", order_id string, line_id string, sku string, quantity long, unit_price_minor long, gross_minor long"
)
REJECT_SCHEMA = SOURCE + ", " + EVIDENCE + ", reasons array<string>"
class DuplicateMember(ValueError):
"""JSON object members must be unique, including inside unknown additive fields."""
class NumericLimit(ValueError):
"""A valid JSON numeric token exceeds the supported representation."""
def bounded_integer(token):
if len(token.lstrip("-")) > MAX_INTEGER_DIGITS:
raise NumericLimit("numeric_representation_limit")
try:
return int(token)
except ValueError:
raise NumericLimit("numeric_representation_limit") from None
def unique_object(pairs):
result = {}
for key, value in pairs:
if key in result:
raise DuplicateMember("duplicate_json_member")
result[key] = value
return result
def reject_constant(value):
raise ValueError("nonfinite_json_number")
def valid_id(value):
return isinstance(value, str) and ID.fullmatch(value) is not None
def parse_record(raw):
"""Validate a whole physical record before any child can enter a clean table."""
result = {
"raw_record_base64": base64.b64encode(raw).decode("ascii"),
"raw_json": None,
"order_id": None,
"customer_id": None,
"currency": None,
"items": None,
"reasons": [],
}
if len(raw) > MAX_RECORD_BYTES:
result["reasons"] = ["record_byte_limit"]
return result
try:
text = raw.decode("utf-8")
except UnicodeDecodeError:
result["reasons"] = ["invalid_utf8"]
return result
result["raw_json"] = text
if not text.strip():
result["reasons"] = ["empty_record"]
return result
try:
value = json.loads(
text,
object_pairs_hook=unique_object,
parse_constant=reject_constant,
parse_float=Decimal,
parse_int=bounded_integer,
)
except DuplicateMember:
result["reasons"] = ["duplicate_json_member"]
return result
except (InvalidOperation, NumericLimit):
result["reasons"] = ["numeric_representation_limit"]
return result
except (ValueError, RecursionError) as error:
result["reasons"] = [
"nonfinite_json_number" if str(error) == "nonfinite_json_number" else "malformed_json"
]
return result
if not isinstance(value, dict):
result["reasons"] = ["top_level_not_object"]
return result
reasons = result["reasons"]
if type(value.get("schema_version")) is not int or value["schema_version"] != 1:
reasons.append("unsupported_schema_version")
for name in ("order_id", "customer_id"):
if valid_id(value.get(name)):
result[name] = value[name]
else:
reasons.append("invalid_" + name)
if value.get("currency") not in ("USD", "GBP"):
reasons.append("unsupported_currency")
else:
result["currency"] = value["currency"]
if "items" not in value:
reasons.append("missing_items")
elif value["items"] is None:
reasons.append("null_items")
elif not isinstance(value["items"], list):
reasons.append("items_not_array")
elif len(value["items"]) > MAX_ITEMS:
reasons.append("item_count_limit")
else:
items, seen = [], set()
for position, item in enumerate(value["items"]):
prefix = f"items[{position}]."
if not isinstance(item, dict):
reasons.append(prefix + "not_object")
continue
for name in ("line_id", "sku"):
if not valid_id(item.get(name)):
reasons.append(prefix + "invalid_" + name)
line_id = item.get("line_id")
if valid_id(line_id):
if line_id in seen:
reasons.append(prefix + "duplicate_line_id")
seen.add(line_id)
quantity, price = item.get("quantity"), item.get("unit_price_minor")
if type(quantity) is not int or not 1 <= quantity <= MAX_QUANTITY:
reasons.append(prefix + "invalid_quantity")
if type(price) is not int or not 0 <= price <= MAX_PRICE:
reasons.append(prefix + "invalid_unit_price_minor")
items.append(
{name: item.get(name) for name in ("line_id", "sku", "quantity", "unit_price_minor")}
)
if not reasons:
result["items"] = items
return result
def arrow_schema():
import pyarrow as pa
item = pa.struct(
[
("line_id", pa.string()),
("sku", pa.string()),
("quantity", pa.int64()),
("unit_price_minor", pa.int64()),
]
)
return pa.schema(
[
("source_id", pa.string()),
("source_sha256", pa.string()),
("line_number", pa.int64()),
("raw_record_base64", pa.string()),
("raw_json", pa.string()),
("order_id", pa.string()),
("customer_id", pa.string()),
("currency", pa.string()),
("items", pa.list_(item)),
("reasons", pa.list_(pa.string())),
]
)
def read_sources(batches):
import pyarrow as pa
schema = arrow_schema()
for batch in batches:
for source in batch.to_pylist():
with Path(source["path"]).open("rb") as stream:
content = stream.read(MAX_FILE_BYTES + 1)
if len(content) > MAX_FILE_BYTES:
raise ValueError("source_file_byte_limit")
if hashlib.sha256(content).hexdigest() != source["source_sha256"]:
raise ValueError("source_file_sha256_mismatch")
# splitlines would treat embedded Unicode/control separators as extra records.
records = content.split(b"\n")
if records[-1] == b"":
records.pop() # the final LF terminates its record, not an extra empty record
if len(records) > MAX_RECORDS_PER_FILE:
raise ValueError("source_file_record_limit")
rows = []
for index, record in enumerate(records, 1):
raw = record + (b"\n" if index < len(records) or content.endswith(b"\n") else b"")
rows.append(
{
"source_id": source["source_id"],
"source_sha256": source["source_sha256"],
"line_number": index,
**parse_record(raw),
}
)
if rows:
yield pa.RecordBatch.from_pylist(rows, schema=schema)
def manifest(paths):
paths = [Path(path) for path in paths]
if len(paths) > MAX_FILES or len({path.name for path in paths}) != len(paths):
raise ValueError("at most 100 source files with unique basenames are supported")
records = []
for path in paths:
with path.open("rb") as stream:
content = stream.read(MAX_FILE_BYTES + 1)
if len(content) > MAX_FILE_BYTES:
raise ValueError("source_file_byte_limit")
record_count = content.count(b"\n") + int(bool(content) and not content.endswith(b"\n"))
if record_count > MAX_RECORDS_PER_FILE:
raise ValueError("source_file_record_limit")
records.append(
{
"source_id": path.name,
"source_sha256": hashlib.sha256(content).hexdigest(),
"path": str(path.resolve()),
}
)
return records
def normalize(spark, parsed):
from pyspark.sql import functions as F
valid = parsed.where(F.size("reasons") == 0)
counts = valid.groupBy("order_id").count().withColumnRenamed("count", "order_copies")
checked = valid.join(counts, "order_id", "inner")
accepted = checked.where("order_copies = 1")
rejects = parsed.where(F.size("reasons") > 0).selectExpr(
*[part.strip().split()[0] for part in REJECT_SCHEMA.split(",")]
)
repeated = checked.where("order_copies > 1").withColumn("reasons", F.array(F.lit("duplicate_order_id")))
rejects = rejects.unionByName(repeated.select(*rejects.columns))
source_columns = ["source_id", "source_sha256", "line_number", "order_id"]
lines = (
accepted.select(*source_columns, F.explode("items").alias("item"))
.select(*source_columns, "item.*")
.withColumn("gross_minor", F.col("quantity") * F.col("unit_price_minor"))
)
totals = lines.groupBy(*source_columns).agg(F.sum("gross_minor").alias("gross_minor"))
orders = accepted.join(totals, source_columns, "left").withColumn(
"item_count", F.size("items").cast("long")
)
orders = orders.withColumn("gross_minor", F.coalesce("gross_minor", F.lit(0).cast("long")))
return {
"orders": orders.selectExpr(*[part.strip().split()[0] for part in ORDER_SCHEMA.split(",")]),
"lines": lines.selectExpr(*[part.strip().split()[0] for part in LINE_SCHEMA.split(",")]),
"rejects": rejects,
}
def temp_dir():
path = Path(tempfile.mkdtemp(prefix="order-events-"))
atexit.register(shutil.rmtree, path, ignore_errors=True)
return path
def readback(spark, frame, path, schema):
import pyarrow.parquet as pq
frame.write.mode("error").parquet(str(path))
if not list(path.glob("*.parquet")):
# Every output is a physically readable typed dataset, including all-reject/empty input.
path.mkdir(parents=True, exist_ok=True)
pq.write_table(frame.limit(0).toArrow(), path / "empty.parquet")
return spark.read.schema(schema).parquet(str(path))
def demo_files(directory, copies=1):
if type(copies) is not int or not 1 <= copies <= 1000:
raise ValueError("copies must be an integer from 1 to 1000")
directory = Path(directory)
directory.mkdir(parents=True, exist_ok=True)
records = (Path(__file__).parent / "inputs/orders.jsonl").read_bytes().splitlines(keepends=True)
paths = []
for group in range((copies + 49) // 50):
path = directory / f"orders-{group:03d}.jsonl"
with path.open("wb") as stream:
for copy in range(group * 50, min(copies, (group + 1) * 50)):
for raw in records:
# Replace only the explicit placeholder in this authored input, retaining malformed records.
stream.write(raw.replace(b"COPY", f"{copy:04d}".encode()))
paths.append(path)
return paths
def run(spark, paths=None, *, copies=1, output_dir=None):
root = temp_dir()
destination = root
if output_dir is not None:
destination = Path(output_dir)
destination.mkdir(parents=True, exist_ok=False)
if paths is None:
paths = demo_files(root / "inputs", copies)
sources = manifest(paths)
inputs = spark.createDataFrame(sources, "source_id string, source_sha256 string, path string")
parsed = inputs.repartition(2).mapInArrow(read_sources, PARSED_SCHEMA)
parsed = readback(spark, parsed, root / "parsed", PARSED_SCHEMA)
outputs = normalize(spark, parsed)
schemas = {"orders": ORDER_SCHEMA, "lines": LINE_SCHEMA, "rejects": REJECT_SCHEMA}
outputs = {
name: readback(spark, frame, destination / name, schemas[name]) for name, frame in outputs.items()
}
input_keys = Counter(
tuple(row) for row in parsed.select("source_id", "source_sha256", "line_number").collect()
)
output_keys = Counter()
for name in ("orders", "rejects"):
output_keys.update(
tuple(row) for row in outputs[name].select("source_id", "source_sha256", "line_number").collect()
)
if input_keys != output_keys or any(count != 1 for count in input_keys.values()):
raise AssertionError("each physical source record must appear in exactly one order or reject row")
return {
name: frame.orderBy("source_id", "line_number", *(["line_id"] if name == "lines" else []))
for name, frame in outputs.items()
}
def parse_cli(argv=None):
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument(
"--copies", type=int, help="independent copies of the authored snapshot, 1..1000 (default: 1)"
)
parser.add_argument(
"--input", type=Path, action="append", help="read a supplied JSONL file; repeat for more files"
)
parser.add_argument("--output-dir", type=Path, help="retain final Parquet datasets in a new directory")
parser.add_argument("--json", action="store_true", help="print output row counts")
args = parser.parse_args(argv)
if args.input is not None and args.copies is not None:
parser.error("--input and --copies cannot be combined")
args.copies = 1 if args.copies is None else args.copies
if not 1 <= args.copies <= 1000:
parser.error("--copies must be from 1 to 1000")
return args
if __name__ == "__main__":
args = parse_cli()
from pysail.spark import SparkConnectServer
from pyspark.sql import SparkSession
server = SparkConnectServer()
server.start()
spark = None
try:
_, port = server.listening_address
spark = SparkSession.builder.remote(f"sc://localhost:{port}").getOrCreate()
outputs = run(spark, paths=args.input, copies=args.copies, output_dir=args.output_dir)
if args.json:
print(json.dumps({name: frame.count() for name, frame in outputs.items()}, sort_keys=True))
else:
for name, frame in outputs.items():
print(name)
frame.show(truncate=80)
finally:
if spark is not None:
spark.stop()
server.stop()
The supported contract
Only integer schema_version: 1 is supported. Missing, null, boolean, string and
other versions are rejected. The required header fields are order_id,
customer_id, currency and items. IDs use 1 to 64 ASCII letters, digits,
underscores or hyphens, beginning with a letter or digit. Currency is explicitly
USD or GBP; no exchange rates or cross-currency totals are computed.
Each child requires line_id, sku, integer quantity from 1 to 1,000,000 and
integer unit_price_minor from 0 to 1,000,000,000. Booleans, numeric strings and
fractional numbers are not coerced to integers. Line IDs must be unique inside
an order; the same SKU on different line IDs is allowed. There are at most 100
children. These limits keep exact integer multiplication and per-order sums
within signed 64-bit range. Zero-price items are accepted.
Missing items, null items, non-array items, invalid children and repeated
child IDs reject the whole order. An empty array is deliberately accepted as an
order with zero children and gross total zero. This is a normalization contract,
not a claim that an order was paid, fulfilled or valid under a business policy.
Duplicate object members at any nesting level, nonstandard NaN/Infinity
constants, malformed JSON, non-object roots, blank records and invalid UTF-8 are
rejected. Numbers beyond the parser's decimal representation produce an explicit
numeric_representation_limit reject, including in unknown fields. Integer
tokens are limited to 4,300 digits before field-specific limits are applied. Validation
emits deterministic ordered reasons for the fields it can
inspect. Syntax, encoding and structural failures produce a single boundary
reason. Unknown additive fields are allowed and retained in the original
evidence; they do not silently become modeled columns. Adding another schema
version requires a new explicit contract.
Separate parent and child tables
Start with normalize() in pipelines/normalize-order-events/main.py for the DataFrame transformation.
run() handles file ingestion and output readback; parse_record() defines the
strict record-validation policy.
- A manifest has a logical filename, SHA-256 and worker-readable path per file.
Each physical LF-delimited record keeps its 1-based line number. LF or CRLF
terminators remain in
raw_record_base64, including malformed records.raw_jsoncontains the original UTF-8 text when decoding was attempted and succeeded. Base64 remains the authoritative byte evidence. - A bounded Python parser validates the complete order inside
mapInArrow. Parsed rows are persisted before later actions. No invalid order contributes a child to the clean tables. - Under this fixed-snapshot policy, every repeated
order_idamong otherwise valid records is quarantined, including exact repeats and repeats across files. Invalid records retain their validation reasons and do not reserve an order ID. This is not replay handling or change-data capture. - The DataFrame transform explodes accepted children once. It aggregates line totals to source-occurrence/order keys before joining them to headers. The left join preserves accepted empty orders without inventing a child. Gross is quantity times unit price only; taxes, shipping and discounts are not included by this contract.
- Orders, lines and rejects are written to separate Parquet datasets and read back with explicit schemas. Typed empty datasets are created when required. The program checks that every physical source occurrence appears exactly once in orders or rejects. Lines reference their accepted parent occurrence.
The source manifest supports at most 100 files with unique basenames, each at most 1 MiB and 10,000 physical records. An unreadable file, file above either limit or changed source SHA fails the job; the program does not pretend it ingested a partial file. Individual records above 64 KiB, including their LF or CRLF terminator when present, are rejected with complete byte evidence, within the file limit. Input paths must be readable by workers. The generated demonstration uses temporary local paths and removes them when the process exits.
This is bounded snapshot normalization, not a general schema-evolution system, transactional publication protocol or distributed storage example. Source data, including unknown fields, is retained in clean-order evidence and reject evidence; apply appropriate access controls and retention to real inputs.
Verification
Maintenance checks compare both engines against an independently assembled reference: authored literal accept/reject classifications, literal expected order and line counts, and separate Python integer arithmetic. They check exact original bytes, additive fields, duplicate barriers, invalid child types, malformed JSON, empty inputs and typed output readback. The reference does not call the production parser to decide whether an input is accepted.