diff --git a/automated_ingestion/sources/san_acacia/dlt_pipeline.py b/automated_ingestion/sources/san_acacia/dlt_pipeline.py index 4c64088f4..465c8729f 100644 --- a/automated_ingestion/sources/san_acacia/dlt_pipeline.py +++ b/automated_ingestion/sources/san_acacia/dlt_pipeline.py @@ -59,6 +59,17 @@ than SO-0125. """ +LOADER_FILE_FORMAT = "parquet" +"""Raw-zone file format. + +dlt writes gzipped JSONL unless told otherwise, and the first live run landed +that way. Parquet is what Mode B replay assumes: replay reads the raw zone +filtered on event time, and a columnar format with real types lets that read a +window without decompressing and parsing every record. It also preserves the +distinction between a null and a missing field, which JSONL round-trips less +reliably. +""" + INITIAL_START = "2015-01-01T00:00:00+00:00" """Floor for a point that has never been ingested. diff --git a/automated_ingestion/sources/san_acacia/ingest.py b/automated_ingestion/sources/san_acacia/ingest.py index 4d7d8a021..ff318d9fc 100644 --- a/automated_ingestion/sources/san_acacia/ingest.py +++ b/automated_ingestion/sources/san_acacia/ingest.py @@ -43,6 +43,7 @@ def _client() -> DiverHubClient: def raw_san_acacia_locations(context: AssetExecutionContext) -> Output[int]: """Land the point roster in the raw zone.""" from automated_ingestion.sources.san_acacia.dlt_pipeline import ( + LOADER_FILE_FORMAT, PROJECT_ID, build_pipeline, vanessen_locations, @@ -51,7 +52,7 @@ def raw_san_acacia_locations(context: AssetExecutionContext) -> Output[int]: client = _client() points = list(client.monitoring_points(PROJECT_ID)) pipeline = build_pipeline() - pipeline.run(vanessen_locations(client)) + pipeline.run(vanessen_locations(client), loader_file_format=LOADER_FILE_FORMAT) context.log.info("landed %s monitoring points", len(points)) return Output( @@ -72,6 +73,7 @@ def raw_san_acacia_locations(context: AssetExecutionContext) -> Output[int]: def raw_san_acacia_readings(context: AssetExecutionContext) -> Output[int]: """Land water levels for every point, isolating per-point failure.""" from automated_ingestion.sources.san_acacia.dlt_pipeline import ( + LOADER_FILE_FORMAT, PROJECT_ID, build_pipeline, vanessen_readings, @@ -86,7 +88,10 @@ def raw_san_acacia_readings(context: AssetExecutionContext) -> Output[int]: pipeline = build_pipeline() failures: list[dict[str, Any]] = [] - info = pipeline.run(vanessen_readings(client, points, end, failures)) + info = pipeline.run( + vanessen_readings(client, points, end, failures), + loader_file_format=LOADER_FILE_FORMAT, + ) rows = _row_count(info) if failures: diff --git a/docs/sources/san_acacia.md b/docs/sources/san_acacia.md index 55725aa40..b184d5636 100644 --- a/docs/sources/san_acacia.md +++ b/docs/sources/san_acacia.md @@ -77,6 +77,22 @@ halve-on-500 recovery is **not** a routine path for `WaterLevels`. Do not assume a 500 there means "too much data" without re-measuring; on `DiverData` that assumption is provably wrong. +## Raw zone format + +Parquet, date-partitioned: + +``` +raw_sanacaciareach/vanessen_readings/year=2026/month=08/day=19/..parquet +``` + +dlt writes gzipped JSONL unless told otherwise, and the first live run landed +that way before this was set. Parquet is what Mode B replay assumes: replay +reads the raw zone filtered on event time, and a columnar format with real types +lets it read a window without decompressing and parsing every record. + +Objects written before this change are `.jsonl.gz`. dlt reads both, so they do +not need migrating, but a replay spanning that boundary reads two formats. + ## Field mapping ### Water levels — the ingested series