Enable Delta optimized writes + lower shuffle partition default

[kathrynalpert]

Description

This is a PR more for comment/visibility than for imminent merge.

Product

N/A

Technical

HL7 ingest workflows of historical data get slower as more days are ingested. On 100k synthetic data, per-workflow duration roughly doubled across 5 IngestHl7ToDeltaLakeWorkflow runs. This PR aims to improve ingest performance and reduce the impact of ingesting into a growing table (what appears to be linear growth).

From Claude:

When the transformer writes its derivative tables (reports_curated, reports_latest, reports_dx, reports_report_patient_mapping), it does so through Spark Structured Streaming — see writeStream.foreachBatch in dataextraction.py:51-62. A single logical "write the new rows" call gets broken into several streaming micro-batches, and each micro-batch goes through a Spark shuffle on the way to disk. Spark's default spark.sql.shuffle.partitions=200 means each tiny micro-batch fans out into up to 200 parquet files. Across one workflow that's hundreds to thousands of tiny files per table; after a few workflows, the count compounds into thousands.

Normally Spark's Adaptive Query Execution (AQE) would catch this — at runtime it would notice that most of those partitions are nearly empty and coalesce them. But AQE is explicitly disabled in Structured Streaming, so the static shuffle.partitions setting is the only knob in this code path.

This PR sets:

• spark.sql.shuffle.partitions: 200 → 32. Spark's 200 default is sized for larger analytical workloads. The derivative tables span a range — reports_latest is on the wide side (a row per latest-version report with the full payload), reports_report_patient_mapping is on the slim side — but in both cases 200 partitions slices the per-shuffle data into chunks too small to be useful, each becoming a tiny output file. 32 keeps shuffles for the wider tables near Spark's recommended ~100-200 MB-per-partition window even at prod's higher per-workflow volume, while still dramatically reducing file count for the slim ones. The transformer's CPU limit doesn't change the analysis much, because the work per partition is small either way.

• spark.databricks.delta.optimizeWrite.enabled: true. Delta's optimized-write feature: before committing a write, Delta does a small repartition to pack rows into fewer, larger files. Acts as a safety net if any individual write would still produce too many small files.

• spark.databricks.delta.autoCompact.enabled: true. After a write, Delta will run a background compaction operation if it sees ≥50 small files in the table. This keeps the file count from drifting upward over many workflows even if individual writes happen to produce a few small files each.

The two spark.databricks.delta.* settings are documented OSS Delta Lake features — see Delta Lake Optimize, which explicitly lists them as Spark SQL session configurations available since Delta 3.1.0. (The databricks. prefix is historical; both settings work in OSS Delta.)

Per-workflow volume varies substantially across environments — dev04 is ~4K HL7 messages per child workflow, prod is ~1.5M (each log file has ~10K messages, batched 150 at a time by hl7log_extractor_concurrency's "continue-as-new" loop). Across that range, the wider derivative tables can push several GB of intermediate shuffle data on a single prod workflow. 32 partitions lands those shuffles in or near Spark's recommended 100–200 MB-per-partition window for prod's wider cases, and is dramatically better than 200 for dev04's smaller workloads. If hl7log_extractor_concurrency itself is significantly raised later, this default should be revisited; cleaner long-term would be for the extractor to pass a derived shuffle.partitions through to the transformer alongside the batch size, but that seemed beyond the scope of this PR that we may not even merge...

Type of change

• Work behind a feature flag
• New feature
• Improvement
• Bug fix
• Refactor (code improvement with no functional changes)
• Documentation update
• Test update

Impact

Security

Authorization

N/A

Appsec

N/A

Performance

Two runs on dev04 against the same daily HL7 batches: a baseline run on the previous defaults (cancelled at WF 6 due to file accumulation and slowdown — see below) and a full 24-workflow run with this PR's settings.

Metric	Baseline (run cancelled at WF 6)	With this PR (24 WFs, run completed)
WF 1 duration	10.66 min	6.09 min
WF 5 duration	20.53 min	13.56 min
WF 6 duration	24.33 min	15.34 min
WF 24 duration	n/a	28.55 min
Per-WF time growth (early, WF 1-14)	~+2.7 min/WF	~+1.5 min/WF
Per-WF time growth (late, WF 15-24)	n/a	flat — duration band 28-32 min
reports_report_patient_mapping files at end of run	1,348 (5 WFs)	86 (24 WFs)
reports_dx files at end of run	999 (5 WFs)	26 (24 WFs)
Median Spark tasks per filter	70	5
Median Spark job duration	30 ms	16 ms

With this PR, the full 24-workflow run completed cleanly: no pod restarts, no Spark errors. File accumulation rates per workflow dropped substantially (mapping table at 86 files after 24 WFs vs 1,348 files after just 5 in the baseline), and per-workflow time growth followed a two-phase pattern: linear ~+1.5 min/WF during the first 14 workflows, then a plateau at 28–32 min from WF 15 through WF 24. Row counts verified via Trino: reports, reports_curated, and reports_latest all hold the expected 100,000 rows — no data loss or duplication.

A caveat on causation: the baseline run was cancelled before it could plateau or fail, so we can't say with certainty that small-file accumulation was the primary driver of its time growth. However, both were observed together, and both improve with this PR's settings. We also can't tell from logs alone whether the plateau in the new run reflects autoCompact engaging (no autoCompact/OptimizeTable entries surfaced, and the file count kept growing past the 50-file threshold rather than dropping) or natural saturation of the patient mapping table as fewer new patients arrive per batch. Either way, the new run reaches a stable steady state rather than degrading indefinitely.

Data

No schema or data-model changes. Optimized writes and auto-compaction are transparent to downstream readers; consumers see the same tables, just backed by fewer, larger files. Compaction is performed by Delta at write time and does not rewrite previously-written data (so existing tables won't be retroactively compacted by this change alone).

Backward compatibility

Compatible: the new settings are session-level Spark configs that apply only to writes performed after the next pod restart. Older Delta files are unaffected. The default change to spark.sql.shuffle.partitions only affects callers that didn't already override spark_sql_shuffle_partitions in their inventory.

Testing

• Manual on dev04: full cold-start (helm uninstall, Delta tables dropped, lake/delta and lake/hl7 wiped, postgres ingest DB dropped) → make install-extractor → ran ingest and recorded per-workflow durations, file counts, and Spark job stats. Verified final row counts.
• The empirical numbers above were measured with spark.sql.shuffle.partitions=16; this PR lands 32 as a safer default for prod's higher per-workflow volume. The two values shouldn't be empirically distinguishable at dev04 scale, and optimizeWrite coalesces the small per-partition data either way.

Note for reviewers

THIS IS A DRAFT, needs validation on preprod.

• This is a targeted spark-defaults change, not a full performance overhaul. Claude cites the remaining source of per-workflow growth as primarily existing_mapping_df scans growing linearly with the mapping table, but I have not explored this claim.

• I upped memory and CPU to better match prod before the first test hl7_transformer_spark_memory: 32G and hl7_transformer_cpu_request: 4, hl7_transformer_cpu_limit: 8

• latesttable.py:43-45 runs OPTIMIZE reports_latest after every workflow's MERGE. With autoCompact now enabled by this PR, that manual OPTIMIZE is largely redundant. It's also the source of some bloat, after the 24-workflow test run, reports_latest holds 1.04 GB on S3 vs reports_curated's 78 MB for the same 100K rows. Claude believes this stems from tombstoned compaction artifacts that the per-workflow OPTIMIZE keeps generating without VACUUM. The queryable content is fine; only S3 storage is bloated. Removing the per-workflow OPTIMIZE (and adding periodic VACUUM regardless) would reclaim the space and simplify the write path.

Checklist

• My code adheres to the coding and style guidelines of the project (and I've run pre-commit run --all-files)
• I have commented my code, particularly in hard-to-understand areas
• I have added or updated user documentation, if appropriate
• I have added or updated technical documentation, including an architectural decision record, if appropriate
• I have added unit tests, if appropriate
• I have added end-to-end tests, if appropriate

[github-actions]

Dependency Review

✅ No vulnerabilities or OpenSSF Scorecard issues found.

Snapshot Warnings

⚠️: No snapshots were found for the head SHA cc01707.

Ensure that dependencies are being submitted on PR branches and consider enabling retry-on-snapshot-warnings. See the documentation for more information and troubleshooting advice.

Scanned Files

None

[johnflavin]

dev05 ingest vs. these metrics

I gathered comparable numbers from a recent 100k-synthetic ingest on the dev05 cluster. TL;DR: the per-workflow durations aren't apples-to-apples with the dev04 numbers above, because dev05 ran all 24 transform workflows concurrently against a 2 GB-memory transformer that OOM-restarted mid-run, vs. dev04's sequential 32 GB run.

What was run

• Parent workflow: IngestHl7LogWorkflow 7a4e188f-a3f2-4a81-b2b7-4187b51b4371, started 2026-05-26 15:54 UTC
• Dataset: /data/100k_varied_oct_29_2025 — the same 100k synthetic dataset
• 3,591 log files, batched 150 at a time → 24 child IngestHl7ToDeltaLakeWorkflow runs (same count and ~4K msgs/child as the dev04 baseline)
• Config: baseline — spark.sql.shuffle.partitions=200, no optimizeWrite, no autoCompact. So dev05 maps to the baseline column, not this PR's proposed settings.
• Resources: Spark driver/executor 2 GB each; transformer pod limit 8 GiB / 4 CPU. (This PR's runs used 32 GB — dev05 had ~1/16th the Spark memory.)

Per-workflow durations (Temporal start→close)

WF (by launch order)	Duration	WF	Duration	WF	Duration
1	684 min	9	158 min	17	1051 min
2	193 min	10	473 min	18	412 min
3	20 min (min)	11	889 min	19	1046 min
4	271 min	12	314 min	20	99 min
5	31 min	13	228 min	21	1062 min
6	45 min	14	1060 min	22	125 min
7	61 min	15	892 min	23	1098 min (max)
8	80 min	16	362 min	24	1053 min

• Min / Median / Max / Mean: 20 / 362 / 1098 / 488 min
• Total wall-clock: 15:54 (05-26) → 10:19 (05-27) ≈ 18.4 hours

Why these aren't apples-to-apples

1. Concurrent, not sequential. The parent continued-as-new 24× in ~6 minutes (15:54–16:00), launching all 24 children nearly at once. They then contended for a single transformer worker. Completion order is uncorrelated with launch order (WF 3 = 20 min, WF 14 = 1,060 min) — the signature of jobs queueing behind one resource-starved worker, not the sequential table-growth-per-run effect this PR measures.
2. Pod OOM-restarts during the run. The transformer restarted 7×, the last at 2026-05-27 09:16 with exitCode 137 (OOM). The final activity ran on attempt 3, so wall-clock durations include retry time after kills. (This PR's run had no pod restarts or Spark errors.)
3. 1/16th the memory (2 GB vs 32 GB), which is almost certainly why dev05 OOM'd under concurrent load where this PR's run was clean.

Net: dev05's per-WF durations reflect queue contention + OOM retries on an under-provisioned, baseline-config box — they can't be placed in the WF1/WF5/WF6/WF24 progression above. The only roughly-comparable headline is total wall-clock (~18.4 h to ingest 100k under these conditions), and even that is confounded by the OOM restarts.

Spark job/task stats — unobtainable

The "median tasks per filter" and "median Spark job duration" rows can't be reproduced for dev05 after the fact:

• spark.eventLog is not enabled and there's no history server → no persisted event logs.
• Spark internal logging is at WARN → no DAGScheduler/TaskSetManager/job-timing lines in pod or Loki logs.
• The per-batch Spark UI was in-process and is gone (pod restarted, run ended 2 days ago).

If we want a genuinely comparable measurement from dev05, the cleanest path is a fresh ingest that mirrors this PR's methodology — sequential workflows, matched Spark memory, and spark.eventLog.enabled=true so the job/task stats are actually captured.