schema-sanitizer

schema-sanitizer turns inconsistent CSV, JSON, JSON arrays, JSON Lines, NDJSON, XML, and Parquet data into stable analytical tables or cleaned files. A native C++23 engine performs schema inference, reconciliation, bounded streaming, and Arrow C Data materialization; the Python API provides file, dataframe, partitioned pipeline, and BigQuery integration helpers.

Version 0.3.6 is still alpha software, with particular focus on Parquet files used by BigQuery external tables.

Documentation

• This README is the installation, Python API, options, and pipeline guide.
• heuristics.md explains inference, field sanitization, schema merging and versioning, the registry, drift records, and the BigQuery sidecar model.
• responsibilities.md maps the Python and C++ source layout for contributors.

Install

Install the core package plus the output adapter you need:

pip install 'schema-sanitizer[pyarrow]'

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

Quick start

import schema_sanitizer as ss

result = ss.to_pyarrow(
    "raw/events.jsonl",
    input_format="jsonl",
    parse_integers=True,
    parse_iso_timestamps=True,
)

table = result.clean_data
print(table.schema)
print(result.schema_drifts)

Write the same cleaned data without retaining an output table in memory:

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

Every conversion returns a schema_sanitizer.Result.

Python API

All public converters are named to_* and share the cleaning options described below.

new_schema_registry() creates an empty registry for a pipeline without depending on the registry JSON structure:

registry = ss.new_schema_registry()

Inputs and formats

input_format is mandatory. It is never inferred from the extension or file contents, and None or "auto" is rejected.

input_mode="single_file" processes exactly one file. input_mode="directory" processes matching direct children in deterministic filename order; it does not recurse into subdirectories.

result = ss.to_pandas(
    "raw/2026-07/",
    input_format="jsonl",
    input_mode="directory",
)

Generated ETL columns

Every analytical and file conversion adds four top-level columns. They always occupy the end of the Arrow schema, physical output file, and generated BigQuery external-table schema in this exact order, regardless of column_order:

1. schema_registry
2. schema_drifts
3. source_file
4. ingestion_timestamp

schema_registry contains the updated canonical registry as JSON and schema_drifts contains this run's drift events as JSON. They are populated on the first output row and null on later rows. source_file and ingestion_timestamp are populated on every row. ingestion_timestamp uses Arrow/Parquet TIMESTAMP_MICROS.

Source fields that use one of these reserved root names are rejected rather than allowed to replace the generated fields. See heuristics.md.

Options

This is the complete option set accepted by the seven public converters. File converters additionally require output_path; to_parquet also accepts its two output compression options.

Paths, selection, and schema

String scalar parsing

These options affect strings such as CSV cells, XML text, and quoted JSON values. JSON numbers and booleans are already typed by JSON syntax. Parsing is opt-in, and unmatched strings remain strings.

String parsers first try the exact value, then retry after trimming surrounding ASCII whitespace. A failed parse preserves the original string and whitespace. When integers and floats coexist, inference promotes the field to float64.

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

Source parsing, errors, and resources

Parquet output

These options are accepted only by to_parquet:

ss.to_parquet(
    "raw/events.jsonl",
    "silver/events.parquet",
    input_format="jsonl",
    parquet_compression="gzip",
    parquet_gzip_level=6,
)

Incremental schemas

Pass one result's registry into the next conversion to preserve schema history:

first = ss.to_parquet(
    "raw/2026-07-12/events.jsonl",
    "silver/2026-07-12/events.parquet",
    input_format="jsonl",
)

second = ss.to_parquet(
    "raw/2026-07-13/events.jsonl",
    "silver/2026-07-13/events.parquet",
    input_format="jsonl",
    schema_registry=first.schema_registry,
    schema_mode="additive",
)

Use schema_mode="strict" when a non-empty existing registry is mandatory and unexpected fields should fail. Detailed compatibility, version-family, and generation behavior is documented in heuristics.md.

Partition pipeline

schema_sanitizer.pipeline provides reusable single-writer building blocks for Hive-style daily or hourly pipelines:

from datetime import date

import schema_sanitizer as ss
from schema_sanitizer.pipeline import (
    HiveRangeConfig,
    build_hive_range_plan,
    discover_existing_source_plans,
    run_partitioned_to_parquet,
)

plans = build_hive_range_plan(
    HiveRangeConfig(
        source_prefix="gs://bronze/events",
        output_prefix="gs://silver/events",
        start_date=date(2026, 7, 1),
        end_date=date(2026, 7, 13),
        input_format="jsonl",
        input_mode="single_file",
        file_name_prefix="events",
    )
)

discovery = discover_existing_source_plans(plans, input_format="jsonl")
pipeline_result = run_partitioned_to_parquet(
    discovery.existing_plans,
    initial_schema_registry=ss.new_schema_registry(),
    to_parquet_kwargs={
        "input_format": "jsonl",
        "schema_mode": "additive",
        "parse_integers": True,
        "parse_iso_timestamps": True,
    },
)

The runner carries the registry returned by each successful partition into the next one. infer_warm_up_schema_registry* can scan a separate range additively before normal writes. Source discovery, warm-up, and writing support the same local and remote paths as the public converters.

The complete production-shaped example is examples/example_07/07_gcs_jsonl_to_silver_parquet_range_prefix.py. It includes daily/hourly planning, directory inputs, missing-partition skips, warm-up, BigQuery registry bootstrap, external-table creation, and sidecar updates.

BigQuery external tables

schema_sanitizer.integrations.bigquery translates a final PyArrow schema into explicit BigQuery external-table DDL. It removes fields supplied by Hive path partitioning and preserves the physical root order, including the four ETL columns at the end.

from schema_sanitizer.integrations.bigquery import (
    BigQueryTableRef,
    ExternalTableSpec,
    external_table_ddl,
)

table_ref = BigQueryTableRef("my-project", "analytics", "events")
spec = ExternalTableSpec(
    source_uris=["gs://silver/events/*"],
    hive_uri_prefix="gs://silver/events",
    partition_columns=(
        ("year", "INT64"),
        ("month", "INT64"),
        ("date", "DATE"),
    ),
)

ddl, skipped_partition_fields = external_table_ddl(
    table_ref,
    final_arrow_schema,
    spec,
)

The integration also exposes ADBC-backed helpers to validate an existing external table, retrieve its latest embedded schema_registry, and execute create/replace DDL. BigQuery and Arrow ADBC connections are supplied by the application; they are not hidden global clients.

Registry sidecar table

Scanning a large external table solely to find its latest registry can be expensive. The optional sidecar is a native BigQuery table with one pointer per external table:

external_table_name STRING NOT NULL
last_ingested_partition STRING NOT NULL

The partition value uses Hive key order, for example:

year=2026/month=07/date=2026-07-13
year=2026/month=07/date=2026-07-13/hour=08

On bootstrap, fetch_latest_schema_registry can use that pointer to query one partition. Missing, invalid, non-native, empty, or failed sidecar lookups fall back to scanning the external table. After the Parquet outputs and external table are successfully updated, update_registry_sidecar_table creates the sidecar if needed and performs an idempotent MERGE.

The sidecar stores only the lookup pointer; the authoritative registry remains the schema_registry value embedded in the output data. See heuristics.md.

Local and cloud filesystems

Local paths, file://, gs:///gcs://, s3://, common Azure Blob/ABFS URIs, and single-file HTTP(S) sources are supported. Install cloud clients with:

pip install 'schema-sanitizer[cloud]'

Remote inputs are staged through provider-native async clients into replayable local temporary files. File outputs are uploaded after conversion. Remote directory listing is bounded, deterministic, and non-recursive; generic HTTP directory listing is not supported.

Useful tuning variables are SCHEMA_SANITIZER_ASYNC_CONCURRENCY (default 64), SCHEMA_SANITIZER_ASYNC_PREFETCH_FILES (default twice the concurrency), SCHEMA_SANITIZER_ASYNC_TIMEOUT (120 seconds), SCHEMA_SANITIZER_ASYNC_RETRIES (4), and SCHEMA_SANITIZER_SPOOL_DIR (the system temporary directory).

Development

Install development dependencies and compile the editable native extension:

python -m pip install -e '.[dev]'

Build the standalone CMake target:

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

Run checks:

python -m pytest -q
python -m ruff check .
python -m mypy
pre-commit run --all-files

Run the end-to-end benchmark suite with a small smoke workload or a focused case:

python benchmarks/bench_ingest.py --rows 100 --width 4 --repeats 1
python benchmarks/bench_ingest.py --case jsonl --rows 100000 --repeats 3

For architecture and ownership, see responsibilities.md. For the production-readiness roadmap, see todo.md.

License

Apache License 2.0. See LICENSE.