schema-sanitizer

Version 0.2.1: this project is still being tuned and tested, especially for generating Parquet files used by BigQuery external tables.

schema-sanitizer converts messy CSV, JSON, JSON Lines, NDJSON, XML, and Parquet data into stable analytical tables or sanitized files. The native C++23 core handles schema inference, scalar/container reconciliation, field versioning, bounded streaming, and Arrow C Data materialization.

Index

• Install
• Public API
• Input Formats
• Input Mode
• Shared Parameters
• Paths And Input Selection
• Schema And Field Handling
• String Scalar Parsing
• Source-Specific Parsing
• Errors And Resources
• Configuration Examples
• Result
• ETL Generated Columns
• Schema Reconciliation
• Field Names
• Timestamp Precision
• Depth Limits
• Memory Safety And Tuning
• Filesystems
• Example 7
• Development
• License

Install

pip install 'schema-sanitizer[pyarrow]'

Optional analytical targets:

pip install 'schema-sanitizer[pandas]'
pip install 'schema-sanitizer[polars]'
pip install 'schema-sanitizer[duckdb]'
pip install 'schema-sanitizer[all]'

import schema_sanitizer as ss

Public API

All public operations are named to_*.

In-memory analytical functions:

File-to-file functions:

events = ss.to_pyarrow(
    "raw/events.jsonl",
    input_format="jsonl",
)

customers = ss.to_pandas(
    "raw/customers.csv",
    input_format="csv",
)

ss.to_parquet(
    "raw/events.jsonl",
    "silver/events.parquet",
    input_format="jsonl",
)

All seven functions expose the same input and cleaning options. File-to-file functions additionally take output_path.

Input Formats

input_format must always be selected explicitly. The signature default is None, but calling any to_* function with None raises an error. Neither None nor "auto" infers a format from the extension or file contents.

The selected format also validates the source extension. For .json files, choose "json" for one document treated as one row or "json_array" for a top-level array of row objects.

jsonl and ndjson use the same newline-delimited JSON parser. Their only difference is the required extension.

Valid JSONL or NDJSON:

{"a": 1}
{"a": 2}

Valid json_array:

[
  {"id": 1, "name": "Ana"},
  {"id": 2, "name": "Luis"},
  {"id": 3, "name": "Marta"}
]

Every top-level json_array element must be an object. The array is split incrementally into rows instead of being materialized as one nested value.

Passing a mismatched extension fails before ingestion:

# Raises: jsonl requires .jsonl, not .ndjson
ss.to_pyarrow("events.ndjson", input_format="jsonl")

Input Mode

input_mode accepts:

Directory traversal is non-recursive. Files with other extensions and nested directories are ignored.

table = ss.to_pyarrow(
    "raw/2026-01/",
    input_format="jsonl",
    input_mode="directory",
).clean_data

Directory behavior:

• jsonl reads only direct .jsonl children.
• ndjson reads only direct .ndjson children.
• json reads direct .json documents as rows.
• json_array flattens each direct .json array into rows.
• csv removes repeated matching headers and rejects header mismatches.
• xml combines direct .xml documents and requires a compatible root/row tag.
• parquet streams direct .parquet and .pq children.

Directory mode requires an explicit input_format.

Shared Parameters

result = ss.to_pyarrow(
    input_path,
    input_format="jsonl",
    input_mode="single_file",
    schema_mode="additive",
    column_order="alphabetically",
    field_name_policy="lower_alpha",
    timestamp_precision="TIMESTAMP_MICROS",
    parse_integers=False,
    parse_floats=False,
    parse_float_decimal_separator=".",
    parse_float_thousands_separator=",",
    parse_iso_timestamps=False,
    parse_iso_dates=False,
    parse_iso_times=False,
    true_tokens=(),
    false_tokens=(),
    custom_timestamp_patterns=(),
    custom_date_patterns=(),
    custom_time_patterns=(),
    arrow_max_depth=32,
    parquet_max_depth=15,
    scalar_object_key="default_key",
    csv_has_header=True,
    csv_delimiter=",",
    input_text_encoding="utf-8",
    xml_row_tag=None,
    on_error="emit_null_row",
    batch_memory_limit_bytes=None,
    read_chunk_bytes=1024 * 1024,
    schema_registry=None,
)

Paths And Input Selection

Schema And Field Handling

String Scalar Parsing

These options apply to string values such as CSV cells, XML text, and quoted JSON values. Actual JSON numbers and booleans are already typed by JSON syntax and do not depend on these options.

Source-Specific Parsing

Errors And Resources

ISO timestamp, date, and time parsing is opt-in. With all three parse_iso_* flags left at False, ISO-looking source strings remain strings. The custom_*_patterns options are independent: configured custom patterns are still applied even when the corresponding built-in ISO parser is disabled.

Float separator options apply only when parse_floats=True and only to string values, including CSV cells and XML text. Real JSON numbers always use JSON's . decimal syntax. Grouping is strict: the default configuration accepts "1,234.56", while European input can use:

result = ss.to_pyarrow(
    "prices.csv",
    input_format="csv",
    parse_floats=True,
    parse_float_decimal_separator=",",
    parse_float_thousands_separator=".",
)

That configuration accepts "1.234,56" and "1234,56". Grouped sections after the first must contain exactly three digits. In comma-delimited CSV, values containing commas must be quoted.

Configuration Examples

European numeric and semicolon-delimited CSV:

prices = ss.to_pyarrow(
    "prices.csv",
    input_format="csv",
    csv_delimiter=";",
    parse_floats=True,
    parse_float_decimal_separator=",",
    parse_float_thousands_separator=".",
).clean_data

Custom Boolean and temporal strings:

events = ss.to_pandas(
    "events.ndjson",
    input_format="ndjson",
    true_tokens=("yes", "active"),
    false_tokens=("no", "inactive"),
    parse_iso_timestamps=True,
    custom_date_patterns=(r"(\d{4})-(\d{2})-(\d{2})",),
).clean_data

Strict incremental conversion using an existing registry:

result = ss.to_parquet(
    "raw/events.jsonl",
    "silver/events.parquet",
    input_format="jsonl",
    schema_mode="strict",
    schema_registry=previous_result.schema_registry,
    on_error="stop",
)

Memory-first processing of a large directory:

result = ss.to_parquet(
    "raw/2026-01/",
    "silver/2026-01.parquet",
    input_format="jsonl",
    input_mode="directory",
    batch_memory_limit_bytes=64 * 1024 * 1024,
    read_chunk_bytes=256 * 1024,
)

to_csv, to_jsonl, and to_parquet return Result.clean_data is None. Analytical functions return their named in-memory object.

Result

Every public function returns schema_sanitizer.Result.

Analytical and file outputs use the same registry-backed native path, so they produce the same schema and metadata behavior.

ETL Generated Columns

Every analytical and file conversion adds these fixed top-level columns:

These columns contain values only in the first output row. Remaining rows are null to avoid repeating large registry payloads.

The names are part of the ETL output contract and cannot be configured. They are reserved at the top level: conversion fails before writing if the source schema already contains any of them. A nested source key such as payload.source_file is allowed because it does not conflict with the generated top-level columns; normal field-name sanitization still applies to that nested key. Reserved top-level source fields are not renamed or versioned, since silently doing so would make downstream registry discovery ambiguous.

Schema Reconciliation

The embedded schema_registry is the source of truth for incremental processing. Pass the latest registry to the next conversion:

result = ss.to_parquet(
    "raw/2026-01-09/events.jsonl",
    "silver/2026-01-09/events.parquet",
    input_format="jsonl",
    schema_registry=previous_registry,
)

next_registry = result.schema_registry

Before generating a field version, the native merge attempts compatible reconciliation:

• A singleton can be wrapped into an existing list.
• A scalar can be wrapped into an existing struct under default_key.
• Empty objects are treated as null.
• New compatible struct children are added as nullable fields.

Irreconcilable drift creates _vN at the lowest incompatible schema level.

sentiment_analysis: struct<...>
sentiment_analysis_v2: list<struct<
  magnitude: double,
  magnitude_v2: string
>>

The struct-to-list drift versions sentiment_analysis. A later float-to-string drift versions only magnitude; it does not create sentiment_analysis_v3.

Existing exact historical variants are preferred during past-date reprocessing. Otherwise the newest compatible container is evolved recursively. Repeating an already known shape does not increment schema_generation.

Materialization routes one source value to one most-compatible sibling:

• Arrays prefer list variants.
• Numeric values prefer numeric scalar variants.
• Ordinary strings prefer string variants.
• Parse-enabled numeric and temporal strings can target typed variants.

Each drift event receives a native UTC detected_at timestamp. The same timestamp is written to the output file's first-row ingestion_timestamp, so the materialized file and its drift audit events share one conversion time. Source partition identity remains available through source_file and any Hive partition columns, including during historical reprocessing.

Field Names

field_name_policy="lower_alpha" keeps lowercase a-z only. lower_snake keeps lowercase letters, digits, and underscores. preserve keeps source names.

Collisions use deterministic suffixes derived from the original dirty key, so source field order does not change the dirty-key to clean-key mapping.

Timestamp Precision

Accepted values:

• TIMESTAMP_MILLIS
• TIMESTAMP_MICROS (default)
• TIMESTAMP_NANOS

Microseconds are the default because BigQuery external tables support Parquet timestamp micros. BigQuery does not accept Parquet TIMESTAMP_NANOS.

Depth Limits

arrow_max_depth counts struct and list containers. parquet_max_depth counts Parquet/BigQuery RECORD levels; list wrappers do not add a RECORD level.

Over-depth nested values are flattened to string-compatible output rather than allowing unbounded schema expansion.

Memory Safety And Tuning

The pipeline uses replayable streaming sources, bounded inference batches, and streaming file writers. batch_memory_limit_bytes controls the approximate per-batch budget.

Memory-first settings for large files:

ss.to_parquet(
    "raw/large.jsonl",
    "silver/large.parquet",
    input_format="jsonl",
    batch_memory_limit_bytes=64 * 1024 * 1024,
    read_chunk_bytes=256 * 1024,
)

64 * 1024 * 1024 is 64 MiB.

Trade-offs:

• Lower batch_memory_limit_bytes reduces peak memory and may reduce speed.
• Lower read_chunk_bytes reduces transient input buffers and increases read calls.
• Parquet decoding enables threads only when the memory budget is large enough.
• Directory mode processes direct child files incrementally rather than loading the full directory at once.
• CSV directory normalization holds at most one configured-size source file in memory while validating and removing repeated headers.

Filesystems

Input and output paths may be local paths or PyArrow filesystem URIs such as:

file:///data/events.jsonl
s3://bucket/events/2026-01-09/events.jsonl
gs://bucket/events/2026-01-09/events.jsonl
abfs://container/events/2026-01-09/events.jsonl

Directory listing uses the same filesystem and is non-recursive.

Example 7

examples/example_07/07_gcs_jsonl_to_silver_parquet_range_prefix.py implements a single-writer daily GCS JSON-array-to-Parquet pipeline with:

• explicit input_format="json_array"
• .json source extension validation
• integer, float, ISO timestamp, ISO date, and ISO time string parsing enabled
• source existence discovery and missing-date skipping
• embedded registry retrieval through Arrow ADBC
• incremental and random past-date reprocessing
• one final BigQuery external-table create/replace operation

Development

pip install -e .[dev]
pytest

Native build:

cmake -S . -B build/dev -G Ninja -DCMAKE_BUILD_TYPE=Release
cmake --build build/dev

License

Apache License 2.0. See LICENSE.