CensusCountyBusinessPatterns: optimize pipeline runtime and update validation - #2219
Open
kartik-s21 wants to merge 61 commits into
Open
kartik-s21 wants to merge 61 commits into
kartik-s21 wants to merge 61 commits into
Conversation
Code fix unenergy
![gemini-code-assist[bot]](https://avatars.githubusercontent.com/in/956858?s=60&v=4){kind=small}
Contributor
There was a problem hiding this comment.
Code Review
This pull request introduces parallel processing for downloading and processing Census County Business Patterns data, caches the StatVars MCF locally to optimize performance, and removes golden data validation checks. The review feedback highlights several critical improvement opportunities: streaming files directly in main.py to avoid potential Out-Of-Memory (OOM) errors during parallel execution, tracking and failing fast on sharding failures in Step 1 of shard_input_csv.sh, and ensuring proper error propagation inside the split_csv shell function.
kartik-s21
force-pushed
the
fix-census-cbp-validation
branch
from
September 11, 2026 11:07
87e0fc4 to
34050f0
Compare
… non-empty output
balit-raibot
self-requested a review
September 17, 2026 03:23
balit-raibot
approved these changes
Sep 17, 2026
…nd remove debug counter references
…g and validation - Fix [P1] empty shard pattern: add shopt -s nullglob, check for generated shards, and abort with exit 1 if no shards exist. - Fix [P1] zero-observation check: verify observation data exists beyond CSV header via line count > 1. - Fix [P1] operational counters: restore counters directory, pass --output_counters to stat_var_processor.py, clean up on failure, and add counters/* to manifest.json source_files. - Fix [P2] runtime bottleneck: add safe bounded parallelism (default 8 workers) using bash wait -n job pool with fail-fast error propagation. - Fix [P2] shell robustness: add set -euo pipefail, quote all variable expansions in split_csv, and verify split succeeds. - Fix [P3] diagnostic logging: fix error message referencing SHARD_DIR instead of INPUT_DIR.
Comment on lines
+109
to
+116
| if python3 "$STATVAR_PROCESSOR_SCRIPT" \ | ||
| --input_data="$file" \ | ||
| --existing_statvar_mcf=gs://unresolved_mcf/scripts/statvar/stat_vars.mcf \ | ||
| --pv_map="censuscountybusinesspatterns_pvmap.csv" \ | ||
| --config_file="censuscountybusinesspatterns_metadata.csv" \ | ||
| --output_path="$OUTPUT_FINAL_DIR/output_${prefix}" \ | ||
| --counters_print_interval=-1 | ||
| # --output_counters="$DEBUG_DIR/counters_${prefix}" \ # uncomment this line to debug the script like to get the details like memory utlization etc. | ||
| # Add any other required arguments for statvar_processpr.py here \ | ||
| # Run in background | ||
|
|
||
| # Manage parallelism: pause if too many jobs are running | ||
| # We monitor the 'statvar_processpr.py' script's processes. | ||
| # sleep_while_active "$PARALLELISM" "$STATVAR_PROCESSOR_SCRIPT" | ||
| else | ||
| echo "WARNING: No shard files found matching '$INPUT_DIR/*_shard_*.csv' or '$file' is not a regular file." | ||
| echo "Please ensure your 'split_csv.sh' generates files in the '$INPUT_DIR' and follows the '*_shard_*.csv' naming convention." | ||
| --counters_print_interval=-1 \ | ||
| --output_counters="$COUNTERS_DIR/counters_${prefix}.csv" && \ |
Contributor
There was a problem hiding this comment.
[P1] Unverified Parallel Rewrite and Redundant 90 MB GCS Schema Downloads Across Workers
- Finding: In commit
7bb12370,shard_input_csv.shwas updated from sequential execution toPARALLELISM=8background workers, with eachprocess_shardinvocation passing--existing_statvar_mcf=gs://unresolved_mcf/scripts/statvar/stat_vars.mcf(89.88 MiB). Underutil/file_util.py,stat_var_processor.pydownloadsgs://inputs into a new/tmp/tmp*file on every invocation (and leaks the raw file descriptor fromtempfile.mkstemp()). Across ~150 shards (and retries), this triggers >13 GB of redundant concurrent GCS downloads of the exact same 90 MiB file—the same source of transient GCS network/timeout failures that caused shards to fail originally. Furthermore, the Cloud Batch test run cited in the PR description (censuscountybusinesspatterns-samnotra-20260914-043508/ GCS version2026_09_13T21_37_35_219006_07_00) ran on 2026-09-13 against commit19a412b5(sequential execution, beforePARALLELISM=8or"counters/*"were added). - Impact: High risk of GCS throttling, transient download timeouts, or
/tmpdisk pressure when 8 concurrentstat_var_processor.pyworkers repeatedly fetchstat_vars.mcfacross ~150 shards, and the current head commit (7bb12370) has not been validated in a Cloud Batch test run. - Recommendation: Download
gs://unresolved_mcf/scripts/statvar/stat_vars.mcfonce to a local file (e.g.,$SHARD_DIR/stat_vars.mcf) before Step 2 and pass--existing_statvar_mcf="$SHARD_DIR/stat_vars.mcf"tostat_var_processor.py. Then run a Cloud Batch test job on the updated commit and update the PR description with the new job link and GCS version URI (gs://datcom-import-test/...).
Comment on lines
14
to
19
| "source_files": [ | ||
| "gcs_output/source_files/*" | ||
| "gcs_output/source_files/*", | ||
| "counters/*" | ||
| ], | ||
| "import_inputs": [ | ||
| { |
Contributor
There was a problem hiding this comment.
[P2] Manifest Missing Mandatory node_mcf and golden_data/*.csv
- Finding:
manifest.jsonupdatessource_filesto add"counters/*", but omits"golden_data/*.csv"even thoughvalidation_config.jsonconfiguresGOLDENS_CHECKrules referencing../../../../golden_data/golden_summary_report.csvand../../../../golden_data/golden_observations.csv. Additionally,import_inputsonly declarestemplate_mcfandcleaned_csv(gcs_output/output/*.csv) and omits"node_mcf": "gcs_output/output/*.mcf", which is mandatory forstat_var_processor.pyimports so generatedoutput_*_stat_vars.mcffiles are uploaded and resolved. - Impact: Golden CSV files are not archived to
<version>/source_files/in GCS as required by the golden validation guidelines, and any new StatVar MCF definitions emitted bystat_var_processor.pyduring future annual refreshes (e.g., new NAICS codes) will be dropped instead of ingested into the graph, causingExistence_MissingReference_variableMeasuredvalidation failures. - Recommendation: Add
"golden_data/*.csv"undersource_filesand add"node_mcf": "gcs_output/output/*.mcf"underimport_inputs[0]inscripts/census_county_business_patterns/manifest.json.
Contributor
There was a problem hiding this comment.
[P2] Missing Date Freshness Validation and Unfiltered Production Dump in golden_summary_report.csv
- Finding: While
manifest.jsonconfigures"validation_config_file": "validation_config.json", inspecting the latest production summary report (2026_07_05T02_04_28_916907_07_00/input0/genmcf/summary_report.csv) shows 6,138 StatVars with non-uniformMaxDatevalues (2023: 5,988 StatVars;2016: 141;2022: 6;2021: 3 due to historical NAICS revisions). However:validation_config.jsoncontains noSQL_VALIDATORdate freshness rule verifyingMaxDate(e.g., ensuring active/canonical StatVars orMAX(MaxDate)meet the expected2023release year within the 2-year CBP lag), leavingMaxDateunvalidated sincegolden_summary_report.csvexcludesMaxDate.golden_data/golden_summary_report.csvchecks in all 6,138 rows (444 KB) of the productionsummary_report.csv(including >6,000 auto-generateddc/...hash StatVars) instead of filtering againstgs://unresolved_mcf/import_validation/nl_statvars.csv(which matches 71 StatVars), andgolden_data/golden_observations.csvcontains 10,391 rows (142 KB) instead of a concise 50–200 row slice.
- Impact: Silent staleness if the latest year fails to download (especially since
main.py:162-167catches 404/BadZipFileonlatest_yearand continues without error), and excessive repo bloat / brittle golden checks from pinningNumPlacesacross all 6,138 StatVars. - Recommendation:
- Add a scoped
SQL_VALIDATORrule invalidation_config.jsonasserting freshness onMaxDate(e.g., verifying that canonical StatVars such asCount_Establishment_NAICSTransportationWarehousing/Count_Worker_NAICSRetailTradehaveMaxDate >= '2023'). - Regenerate
golden_data/golden_summary_report.csvfiltered againstgs://unresolved_mcf/import_validation/nl_statvars.csv(71 rows) usingvalidator_goldens.py, and keep golden fixtures concise (50–200 rows).
- Add a scoped
Comment on lines
+60
to
+62
| mkdir -p "$OUTPUT_FINAL_DIR" | ||
| mkdir -p "$SHARD_DIR" | ||
| echo "Directories created/ensured: $DEBUG_DIR, $OUTPUT_FINAL_DIR" | ||
| SHARD_ROWS=500000 | ||
| mkdir -p "$COUNTERS_DIR" |
Contributor
There was a problem hiding this comment.
[P2] Uncleaned Working Directories Across Repeated Executions and Delayed Failure Termination
- Finding:
- Lines 60–62 run
mkdir -p "$OUTPUT_FINAL_DIR" "$SHARD_DIR" "$COUNTERS_DIR"without clearing stale*_shard_*.csv,output_*, orcounters_*files from prior executions. Ifshard_input_csv.shis re-run in a persistent environment or after a partial run with fewer shards,shards=("$SHARD_DIR"/*_shard_*.csv)and"cleaned_csv": "gcs_output/output/*.csv"will pick up stale shard artifacts. - In
process_shard(lines 122–123), when attempt 3 of 3 fails, the script still logs"Retrying in 15s..."and sleeps 15 seconds before exiting the loop. Furthermore, when a background worker fails (job_failed=1at line 148), the script breaks out of spawning new jobs but waits inwhile [ "$active_jobs" -gt 0 ]; do wait -n ...for all other active background workers to finish rather than terminating active background jobs immediately.
- Lines 60–62 run
- Impact: Potential ingestion of stale artifacts on repeated runs, and wasted Batch compute time on terminal shard failure.
- Recommendation: Clean
"$SHARD_DIR"/*_shard_*.csv,"$OUTPUT_FINAL_DIR"/output_*, and"$COUNTERS_DIR"/counters_*before Step 1; skip thesleep 15backoff whenattempt -eq 3; and terminate remaining background jobs (kill $(jobs -p) 2>/dev/null || true) whenjob_failed=1.
…nsusCountyBusinessPatterns - Address code review comments: - Remove goldens data and node_mcf configuration for script-based import. - Remove operational counters generation from manifest and sharding script. - Update validation config with SQL date freshness check on canonical StatVars. - Pipeline and Cloud Batch performance optimizations: - Parallelize Census zip downloads in main.py using ThreadPoolExecutor (8 threads). - Stream input files directly to DictReader in main.py, avoiding multi-GB memory buffers. - Parallelize text processing across CPU cores in main.py via ProcessPoolExecutor. - Cache schema MCF locally once in shard_input_csv.sh to eliminate redundant downloads. - Parallelize Step 1 sharding and inject headers via split --filter without sed -i disk rewrites. - Increase shard chunk size to 1M rows and scale PARALLELISM up to 48 workers on 64-core VMs. - Add fail-fast termination on worker job failure in shard_input_csv.sh.
Contributor
Author
|
Thank you for the thorough review. All feedback from Round 2 has been addressed in commit Summary of Fixes
Cloud Batch Verification Results
Please let me know if any further adjustments are needed! |
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
4 participants
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Summary
This PR resolves validation failures, eliminates redundant network calls, and significantly improves runtime performance for
CensusCountyBusinessPatterns.Key Changes
Pipeline & Concurrency Optimization:
gs://unresolved_mcf/scripts/statvar/stat_vars.mcfonce locally before shard processing, eliminating >13 GB of redundant concurrent GCS downloads across parallel workers.split --filterto inject headers directly on write, avoiding expensivesed -iin-place disk rewrites.PARALLELISMup to 48 workers on 64-core VMs.main.py: Switched to streaming decompressed inputs directly intocsv.DictReaderand addedThreadPoolExecutorfor zip downloads, eliminating multi-GB memory buffers and reducing runtime.Fault Tolerance, Cleanup & Error Propagation:
$SHARD_DIR/*_shard_*.csvand$OUTPUT_FINAL_DIR/output_*before Step 1 to prevent ingestion of stale artifacts from prior runs.kill $(jobs -p)) and abort the pipeline.process_shardto skip the 15s delay after the 3rd failed attempt.Validation & Configuration Updates:
check_date_freshness_canonical_statvarstovalidation_config.jsonassertingMaxDate >= '2023'on active canonical StatVars (Count_Establishment_NAICSTransportationWarehousing,Count_Worker_NAICSRetailTrade).manifest.jsonandshard_input_csv.sh. Removed legacy golden data dumps and omittednode_mcfsince all StatVars are canonical/pre-resolved (0 new StatVar MCF nodes emitted).Verification & Artifacts
censuscountybusinesspatterns-samnotra-20260927-032649SUCCEEDED(Runtime: 6h 02m 58s / 21,778s, down from 18+ hours)gs://datcom-import-test/scripts/census_county_business_patterns/CensusCountyBusinessPatterns/2026_09_26T20_29_04_561640_07_00/validation_output.csvdiffer_summary.json(73,457,237 observations, 0 diffs)