Fix epacriteriaairpollutants - #2206
shourya116 wants to merge 12 commits into
Conversation
- Pre-declare 47 unindexed EPA Air Quality monitoring stations in EPA_AirQuality.mcf - Standardize units and add retry logic in air_quality.py - Add node_mcf, validation_config_file, and resource limits to manifest.json - Add validation_config.json with strict zero lint error threshold
…ality.mcf to eliminate remote DC API calls
|
Thanks for your pull request! It looks like this may be your first contribution to a Google open source project. Before we can look at your pull request, you'll need to sign a Contributor License Agreement (CLA). View this failed invocation of the CLA check for more information. For the most up to date status, view the checks section at the bottom of the pull request. |
![gemini-code-assist[bot]](https://avatars.githubusercontent.com/in/956858?s=60&v=4){kind=small}
There was a problem hiding this comment.
Code Review
This pull request introduces a diagnostic runbook, configures validation rules, and updates the air_quality.py script to handle unit mapping, filter out cross-border monitors, and use a robust HTTP session with retries for downloading data. Feedback on the changes includes addressing a critical runtime AttributeError when configuring retries, relaxing the lint error threshold in validation_config.json to prevent pipeline failures from new stations, and resolving a potential KeyError and eager evaluation issue when parsing observation units.
…and clean up validation config
…s-border monitor filtering
…ource_files, add date validation rules, and add unit tests
…force strict zero validation thresholds - Restore all 8,408 EPA monitoring station declarations in EPA_AirQuality.mcf to eliminate remote DC API calls, missing observationAbout errors, and location reference warnings - Restore node_mcf in manifest.json and keep differ disabled - Set strict zero threshold for check_missing_refs_count and check_lint_error_count in validation_config.json - Format air_quality_test.py with yapf
abhishekjaisw
left a comment
abhishekjaisw
left a comment
There was a problem hiding this comment.
Review scope
- Target: PR #2206 (
7cdda42598c5cce772ee173325c655b016f6d083) - Reviewed:
EPA_AirQuality.mcf,air_quality.py,air_quality_test.py,manifest.json,test_data/test_import.csv,validation_config.json - Skipped: None
Summary of Progression
Thank you for updating the PR description with the detailed incident summaries and staging verification (epacriteriaairpollutants-shouryasingh-20260914-075748 -> gs://datcom-import-test/scripts/us_epa/airdata/EPACriteriaAirPollutants/2026_09_14T01_01_32_581690_07_00, achieving ImportStatus.STAGING with 0 missing refs and 0 lint errors), and for adding unit tests (7/7 passing).
Three remaining schema/automation items need attention before merging (see inline comments):
- [P1] Remove duplicate canonical StatVars & Units from
EPA_AirQuality.mcf(lines 1–158): Inspectinginput0/genmcf/instance_mcf_nodes.mcffrom the Sep 14 run confirms thatgenmcfingests all 18 canonical StatVars and 2 units intoinstance_mcf_nodes.mcf(lines 97460 and 114284). Becausegenmcf's built-in schema cache already resolves canonical Google3 StatVars (manual_epa_stat_vars.mcf) and units (unit.mcf) locally withoutnode_mcf, lines 1–158 should be removed so only the 21MeasurementMethodEnumnodes remain inEPA_AirQuality.mcf. - [P1] Dynamically generate
AirQualitySitenodes (EPA_AirQuality_sites.mcf) instead of hardcoding 8,404 stations in Git: Pre-populatingAirQualitySitenodes vianode_mcfis necessary sogenmcfpopulateslocal_nodesbefore validatingtable_mcfrows—preventing remote DC API lookups and achieving0missing refs and0lint errors. However, hardcoding 8,404 static stations in Git paired withthreshold: 0means future automated cron refreshes (15 22 10 1,7 *) will fail whenever EPA adds a single new monitoring station. Havingair_quality.pydynamically write uniqueAirQualitySitenodes toEPA_AirQuality_sites.mcfat runtime and setting"node_mcf": "EPA_AirQuality*.mcf"inmanifest.jsonpreservesthreshold: 0automatically on every future refresh without 50,000+ lines of static Git bloat. - [P2] Use dynamic year expression in
check_date_freshness: Replace static'2024'withCAST(SUBSTR(latest_date, 1, 4) AS INT) >= YEAR(CURRENT_DATE) - 1.
Positive findings
scripts/us_epa/airdata/air_quality.py:221-236- Resilient HTTP Session with Exponential Backoff and Retries ✓- Finding: Good - Configuring
requests.Sessionwithurllib3.util.Retry(10 retries, backoff factor 2, status forcelist[429, 500, 502, 503, 504]) and 120s timeout provides strong fault tolerance across 270+ annual archive downloads.
- Finding: Good - Configuring
scripts/us_epa/airdata/air_quality.py:138-140&air_quality_test.py:63-151- Foreign Monitor Filtering and Unit Mapping with Unit Tests ✓- Finding: Good - Filtering out cross-border monitors (
State Codein('80', 'CC')) and normalizing unit strings (UNIT_MAP) cleanly resolves 875K fatalobservationAbouterrors, verified by unit tests (7/7passing).
- Finding: Good - Filtering out cross-border monitors (
scripts/us_epa/airdata/manifest.json:25-33- Appropriate Resource Sizing, Differ Bypass, and PR Documentation ✓- Finding: Good - Allocating 32 vCPUs / 128 GiB RAM / 512 GB disk, setting
invoke_differ_tool: false, and documenting the RCA and staging proof in the PR description prevents the 121.83 GB MCF in-memory diffing OOM kernel freeze.
- Finding: Good - Allocating 32 vCPUs / 128 GiB RAM / 512 GB disk, setting
Coverage
| File | Status | Result |
|---|---|---|
| scripts/us_epa/airdata/EPA_AirQuality.mcf | Reviewed | Two P1 findings |
| scripts/us_epa/airdata/manifest.json | Reviewed | Two P1 findings |
| scripts/us_epa/airdata/validation_config.json | Reviewed | One P1 finding, One P2 finding |
| scripts/us_epa/airdata/air_quality.py | Reviewed | Positive findings |
| scripts/us_epa/airdata/air_quality_test.py | Reviewed | Positive findings |
| scripts/us_epa/airdata/test_data/test_import.csv | Reviewed | No findings |
Verification and limitations
- Checks run: Executed all 7 unit tests in
scripts/us_epa/airdata(7/7 OK) and verifiedyapfformatting (0 diffs) in detached worktree at7cdda42598c5cce772ee173325c655b016f6d083; audited GCS staging artifacts (instance_mcf_nodes.mcf,report.json,import_summary.json) for2026_09_10T13_01_42_117377_07_00and2026_09_14T01_01_32_581690_07_00; verified Google3 schema declarations via Code Search. - Checks not run: Live download of 1980–2025 EPA AQS archives (~3.7h runtime).
- Limitations: Note that inline comments for
EPA_AirQuality.mcfare attached tomanifest.jsonandvalidation_config.jsonbecauseEPA_AirQuality.mcf(+50,525 lines) exceeds GitHub's diff size limit for inline comments.
| "template_mcf": "EPA_AirQuality.tmcf", | ||
| "cleaned_csv": "EPA_AirQuality.csv" | ||
| "cleaned_csv": "EPA_AirQuality.csv", | ||
| "node_mcf": "EPA_AirQuality.mcf" |
There was a problem hiding this comment.
[P1] Redundant declaration of existing canonical StatVars and Units in EPA_AirQuality.mcf (node_mcf) causes double-ingestion as instance nodes
(Comment attached to manifest.json because EPA_AirQuality.mcf exceeds GitHub's diff size limit for inline comments.)
Finding: EPA_AirQuality.mcf lines 1–146 declare 18 StatisticalVariable nodes (Mean_Concentration_AirPollutant_Ozone, etc.) that already exist in canonical Google3 schema (//depot/google3/third_party/datacommons/schema/stat_vars/manual_epa_stat_vars.mcf), and lines 147–158 declare 2 UnitOfMeasure nodes (MicrogramsPerCubicMeter_lc, MicrogramsPerCubicMeter_25C) that already exist in //depot/google3/third_party/datacommons/schema/core/unit.mcf. Because air_quality.py is a non-StatVarProcessor import and manifest.json includes "node_mcf": "EPA_AirQuality.mcf", genmcf ingests all 18 canonical StatVars and 2 units directly into input0/genmcf/instance_mcf_nodes.mcf (verified at lines 97460 and 114284 in the Sep 14 staging run). Note that genmcf's built-in schema cache already resolves these canonical StatVars and units locally without node_mcf (as shown by the Sep 10 run reporting 0 missing reference errors on variableMeasured or unit).
Impact: Causes double-ingestion of 18 canonical StatVars and 2 units as instance nodes in the Knowledge Graph.
Recommendation: Remove lines 1–158 (the 18 StatisticalVariable nodes and 2 UnitOfMeasure nodes) from EPA_AirQuality.mcf, retaining only the 21 MeasurementMethodEnum nodes (Ozone_8hour_2015, NO2_1hour_2010, etc., lines 159–244) which do not exist in Google3 schema.
| "rule_id": "check_missing_refs_count", | ||
| "validator": "MISSING_REFS_COUNT", | ||
| "params": { | ||
| "threshold": 0 |
There was a problem hiding this comment.
[P1] Fragile static snapshot of 8,404 AirQualitySite nodes in EPA_AirQuality.mcf paired with threshold: 0 validation
Finding: Checking 8,404 hardcoded AirQualitySite instance nodes (lines 246–50671 of EPA_AirQuality.mcf) into Git pre-populates genmcf's local symbol table (local_nodes), which resolved the observationAbout lint errors and location warnings seen when node_mcf was omitted (because genmcf does not index streamed table_mcf entities into local_nodes prior to reference checking). However, hardcoding a static station list in Git paired with threshold: 0 on check_missing_refs_count and check_lint_error_count is operationally brittle: whenever EPA commissions even 1 new monitoring station in future scheduled cron runs (cron_schedule: "15 22 10 1,7 *"), that station will be absent from EPA_AirQuality.mcf, causing genmcf to query the remote DC API (where the new station does not yet exist) and emit missing reference errors/warnings that immediately fail validation against threshold: 0.
Impact: Future automated cron refreshes will break whenever upstream EPA adds a new air quality monitoring station, requiring manual code updates to a 50,000-line file, and repository history is bloated by 1.37 MB of instance data.
Recommendation:
- Update
air_quality.py(write_csv) to dynamically write uniqueAirQualitySitenodes encountered during CSV processing to a generated fileEPA_AirQuality_sites.mcfat runtime. - Remove the 8,404 hardcoded station nodes (lines 246–50671) from
EPA_AirQuality.mcf. - Update
manifest.jsonto use a wildcard pattern"node_mcf": "EPA_AirQuality*.mcf"sogenmcfloads bothEPA_AirQuality.mcf(the 21 measurement methods) and the dynamically generatedEPA_AirQuality_sites.mcfon every run.
| "validator": "SQL_VALIDATOR", | ||
| "params": { | ||
| "query": "SELECT MAX(MaxDate) AS latest_date FROM stats", | ||
| "condition": "latest_date >= '2024'" |
There was a problem hiding this comment.
[P2] Hardcoded static year cutoff (>= '2024') in check_date_freshness
Finding: check_date_freshness uses a static string literal "condition": "latest_date >= '2024'". Because air_quality.py dynamically processes data up to datetime.now().year - 1 (MaxDate is 2025-12-31 in 2026), a hardcoded '2024' threshold will never advance in future years (2027, 2028, etc.), allowing multi-year upstream data stalls to pass validation undetected.
Impact: Future scheduled refreshes that fail to ingest new annual files will pass freshness validation silently.
Recommendation: Use a dynamic DuckDB date expression verifying that MaxDate is within the expected 1-year publication lag:
"condition": "CAST(SUBSTR(latest_date, 1, 4) AS INT) >= YEAR(CURRENT_DATE) - 1"3 participants
Summary
Resolves two consecutive execution failures of the
EPACriteriaAirPollutantsimport pipeline:ImportStatus.VALIDATIONdue to 18.1M missing measurement method references (NO2_1hour_2010), 875K fatalobservationAboutreference errors from an unregistered Mexican cross-border monitor (epa/800060001), and differ summaryDATA_ERROR.mcf_file_util.load_mcf_nodes) attempting to load 121.83 GB of MCF text (305M nodes) into Python dictionaries on a 128 GiB RAM VM.Key Changes
1. Data Extraction & Pipeline Reliability (
scripts/us_epa/airdata/air_quality.py)'80'for Mexico and'CC'for Canada) that lack Data Commons place entities, eliminating ~875K fatal lint errors.UNIT_MAP(MicrogramsPerCubicMeter_lc,MicrogramsPerCubicMeter_25C,PartsPerMillion,PartsPerBillion) with fallback to camel case.requests.Session()withurllib3.util.Retry(10 retries, exponential backoff factor 2, retry status codes[429, 500, 502, 503, 504], and 120s timeout) to prevent dropped connections during multi-year archive extraction.2. Schema Declarations & Zero-Error Pre-declarations (
scripts/us_epa/airdata/EPA_AirQuality.mcf)NO2_1hour_2010,Ozone_8hour_2015,SO2_1hour_2010,PM25_24hour_2012,PM10_24hour_2006, etc.).MicrogramsPerCubicMeter_lc,MicrogramsPerCubicMeter_25C).AirQualitySite) to satisfy strict zero lint errors and zero missing references in local resolution mode.3. Automation & Differ Guardrail (
scripts/us_epa/airdata/manifest.json)"config_override": {"invoke_differ_tool": false}to safely bypass the in-memory differ on the 121+ GB MCF output and prevent VM memory exhaustion."validation_config_file": "validation_config.json"."node_mcf": "EPA_AirQuality.mcf"underimport_inputs.cpu: 32,memory: 128,disk: 512).4. Import Validation Configuration (
scripts/us_epa/airdata/validation_config.json)check_deleted_records_percentto preventDATA_ERRORwhen the differ tool is bypassed.threshold: 0forcheck_missing_refs_countandcheck_lint_error_count.check_max_date_consistent(MAX_DATE_CONSISTENT): Verifies uniform MaxDate across all 18 air quality StatVars.check_date_freshness(SQL_VALIDATOR): Enforces that maximum observation date meets freshness requirements (latest_date >= '2024').5. Unit Tests & Test Data (
scripts/us_epa/airdata/air_quality_test.py,test_data/test_import.csv)test_filter_cross_border_monitors).test_unit_mapping).Verification and Test Run Proof
Cloud Batch Staging Execution
gs://datcom-import-test/scripts/us_epa/airdata/EPACriteriaAirPollutants/2026_09_14T01_01_32_581690_07_00Exit Code: 0)ImportStatus.STAGING.Validation Report (
validation_output.csv)check_empty_importcheck_missing_refs_countthreshold: 0)check_lint_error_countthreshold: 0)check_max_date_consistent2025-12-31across all 18 StatVarscheck_date_freshnesslatest_date >= '2024'satisfied (2025-12-31)Local Unit Tests
python3 -m unittest discover -v -s scripts/us_epa/airdata -p "*_test.py": 7/7 tests passed.References