USFEMA_FloodInsuranceClaims: Optimize download and transformation pipeline

[kartik-s21]

Summary

Optimizes and stabilizes the USFEMA_FloodInsuranceClaims data pipeline to eliminate multi-hour runtime bottlenecks, enhance download resilience against API snapshot discrepancies, ensure atomic file publication, and resolve all review and Code Review Agent (CRA) findings.

Script execution time has been reduced from ~61 hours (~219,660s) to ~78.6 seconds, processing 13,765,520 rows with zero missing references and zero lint errors.

---

Key Changes

1. Download Stage (fema_download.py)

• Bulk Streaming & Resilience: Implements fast streaming from OpenFEMA's static bulk CSV endpoint (~60s) while retaining automatic fallback to paginated API requests.
• Retry & Exponential Backoff for API Counter: Wrapped get_total_records() with _retry_method (tries=5, delay=5, backoff=2) from util.download_util_script to prevent transient network timeouts or HTTP 5xx errors from rejecting a valid ~3 GB bulk download and triggering an expensive 2.5-hour pagination fallback.
• Optimized Metadata Query Payload: Added $top=1 to the count query (?$top=1&$count=true), reducing the returned payload from 1,000 serialized JSON records to 1 sample record for faster latency.
• Bulk Tolerance Margin: Added a 2% tolerance ratio (tolerance_ratio=0.98) when comparing bulk CSV row counts against the live OpenFEMA API $count=true counter. This prevents false-negative fallbacks to a 2.5-hour pagination loop caused by periodic snapshot lags.
• Single Live Counter Query: Eliminated redundant get_total_records() API calls by caching and reusing the live record count between bulk validation and pagination fallback.
• Atomic Publication: Added _publish_file_atomically using staging files and os.replace with errno.EXDEV fallback (shutil.move) to protect against corrupted or partial downloads.
• Error Handling Architecture: Replaced logging.fatal() prior to raise with logging.error(...) + raise across library functions, reserving logging.fatal() for top-level CLI termination in main(argv).

2. Processing Stage (process.py)

• Multi-Process Vectorization: Replaced sequential iteration with a multi-process vectorized pipeline using concurrent.futures.ProcessPoolExecutor (multiprocessing.get_context('spawn')) with dynamic core scaling.
• FIPS Code Leading Zeroes: Fixed missing leading zeroes for FIPS codes < 10 (e.g. California, Connecticut, Alabama), preserving 100% of sub-state county and census tract entities.
• Atomic File Writes: Enforced atomic .tmp staging and replace for all output .csv, .tmcf, .mcf, and counters.
• Safe Parsing: Replaced eval() with ast.literal_eval() for safe mapping configuration ingestion.
• Library vs CLI Error Logging: Cleanly separated modular library exceptions (logging.error + raise) from CLI process termination (logging.fatal in main(_)) to resolve dead code and ensure test compatibility.

3. Test Infrastructure & CI (fema_download_test.py, process_test.py, util/download_util_script.py)

• Direct Invocation: Injected _SCRIPT_DIR into sys.path across test modules to support direct execution from the repository root via python3 -m unittest.
• CodeQL URL Sanitization: Sanitized mock test URLs with trailing slashes to eliminate CodeQL alert warnings.
• Test Coverage: Added test cases for bulk download tolerance validation, single API counter querying, external CWD path resolution, and _retry_method HTTP retry/backoff validation. All 29 unit tests pass cleanly.
• CI Formatting: Fixed 1-space indentation error on req_lib.head in util/download_util_script.py to fix presubmit CI.

4. Manifest & Validation (validation_config.json, manifest.json, README.md)

• Date Freshness Validation: Added scoped MAX_DATE_LATEST rule (check_max_date_latest_aggregates) targeting the 6 primary aggregate and active flood risk-zone StatVars (CountOfClaims & SettlementAmount for Building & Contents, High Risk, and Low Risk) to guarantee current-year observations are verified on automated pipeline runs without false alarms on historical deprecated flood zone sub-codes.
• Removed obsolete golden fixtures and configured check_deleted_records_percent with threshold: 0.1 (0.1%), fully conforming to Data Commons guidelines.
• Updated node_mcf to "output/*.mcf" and included "manifest.json" in source_files.
• Expanded README.md with layout, CLI usage, and test instructions.

---

Test Environment & Latest Job Runs

• Cloud Batch Job Run (Latest): Job Run Details
• Validation Output: validation_output.csv

Performance Benchmarks

• Script Execution Time: Reduced from ~219,660s (~61 hrs) to 78.58s
• GenMCF Execution Time: 831.2s
• Total Rows Processed: 13,765,520

---

Documentation & Reports

• Bug: b/507397157
• CRA Review Report: paste/6536877964918784
• Postmortem Report: paste/6373962741841920
• Diagnostics & Performance Optimization Report: paste/5876880071327744

[gemini-code-assist]

Code Review

This pull request re-engineers the FEMA flood insurance claims import pipeline, replacing a slow, sequential, row-by-row legacy processor with a highly optimized, vectorized, multi-process chunked pipeline. It introduces direct bulk CSV downloading with paginated API fallback, updates the manifest and documentation, and adds comprehensive unit tests. Feedback suggests further optimizing the generation of StatVar names in process.py by replacing .apply() with .map() using pre-computed unique flood zone names to avoid Python function call overhead when generating millions of observations.

[SandeepTuniki]

[P2] Missing explicit date freshness validation rule in validation_config.json

Finding: validation_config.json removes check_goldens_summary_report and Check_goldens_output_csv, leaving only check_deleted_records_percent (threshold: 0.1) without any date freshness validation rule (MAX_DATE_LATEST or SQL_VALIDATOR). Inspection of MaxDate across the 309 StatVars in summary_report.csv shows differing MaxDate values across historical flood zone sub-codes (e.g., FEMAFloodZoneAA / FEMAFloodZoneAS), while primary aggregate and active flood risk-zone series always contain current-year observations (2026-05 / 2026-06).

Impact: Without a scoped MAX_DATE_LATEST or SQL_VALIDATOR freshness rule in validation_config.json, stale upstream snapshots or regressions that drop recent years/months will pass automated validation undetected.

Recommendation: Add a scoped MAX_DATE_LATEST (or SQL_VALIDATOR) rule in validation_config.json targeting the primary aggregate and active risk-zone StatVars:

{
    "rule_id": "check_max_date_latest_aggregates",
    "description": "Verifies that core aggregate NFIP claim StatVars contain current-year data.",
    "validator": "MAX_DATE_LATEST",
    "scope": {
        "variables": {
            "dcids": [
                "CountOfClaims_NaturalHazardInsurance_BuildingStructureAndContents_FloodEvent",
                "SettlementAmount_NaturalHazardInsurance_BuildingStructureAndContents_FloodEvent",
                "SettlementAmount_NaturalHazardInsurance_BuildingStructure_FloodEvent",
                "SettlementAmount_NaturalHazardInsurance_BuildingContents_FloodEvent",
                "CountOfClaims_NaturalHazardInsurance_FEMAHighRiskFloodZone_BuildingStructureAndContents_FloodEvent",
                "SettlementAmount_NaturalHazardInsurance_FEMALowRiskFloodZone_BuildingStructureAndContents_FloodEvent"
            ]
        }
    },
    "params": {}
}

[kartik-s21]

Added the scoped MAX_DATE_LATEST rule (check_max_date_latest_aggregates) targeting the 6 primary aggregate and active flood risk-zone StatVars (CountOfClaims & SettlementAmount for Building & Contents, High Risk, and Low Risk).

[SandeepTuniki]

[P2] Un-retried get_total_records API call in fema_download.py can discard a valid 3 GB bulk download

Finding: get_total_records(api_url) executes a single requests.get(url, timeout=30) with no retry/backoff logic. In download_data(), get_total_records(api_url) is invoked after downloading the ~2.98 GB bulk CSV (FimaNfipClaims.csv). If that single metadata API call experiences a transient network or HTTP 5xx error, the except block sets total_records = None, rejects the already-downloaded 2.98 GB bulk file, deletes temp_dir in finally, and falls back to the 2,722-page pagination loop (calling get_total_records(api_url) a second time at line 248, which will abort the entire job if the transient error persists).

Impact: A single transient failure on the lightweight $count=true API endpoint either forces an unnecessary ~2.5-hour paginated download after already streaming 2.98 GB of valid bulk data, or causes the entire import job to fail.

Recommendation: Wrap the HTTP request inside get_total_records(api_url) with bounded retries and exponential backoff (for example, reusing _retry_method from download_util_script.py):

from download_util_script import download_file, _retry_method

def get_total_records(api_url):
    """Makes a preliminary API call to get the total number of records."""
    url = f"{api_url}?$top=1&$count=true"
    try:
        response = _retry_method(url, headers=None, tries=5, delay=5, backoff=2)
        data = response.json()
        ...

[kartik-s21]

Updated get_total_records to use _retry_method from download_util_script.py with 5 retries and exponential backoff (delay=5, backoff=2) to protect against transient metadata API hiccups. Also added $top=1 to the query URL (?$top=1&$count=true) so only 1 sample record is streamed in the payload rather than the default 1,000 records. All unit tests updated and passing.