USFEMA_FloodInsuranceClaims

[kartik-s21]

Description

This PR optimizes and stabilizes the data download pipeline for the FEMA NFIP Flood Insurance Claims import.

Key Changes

1. Direct Bulk Download Support:

   • Added bulk_url support to download the full dataset (FimaNfipClaims.csv) directly from OpenFEMA as the primary method, significantly speeding up the download process.
   • Implemented automatic fallback to paginated API download if direct bulk download fails or returns empty.

2. Error Handling & Fallback Visibility:

   • Explicitly raise exceptions upon bulk download failure so that fallback events are properly captured and logged with warnings.
   • Added fatal logging and exceptions on API chunk failures or incomplete downloads to prevent silent partial data ingestion.

3. CSV Chunk Concatenation Fixes:

   • Improved binary chunk merging logic to ensure consistent newline endings across chunks.
   • Guarded chunk concatenation to prevent empty/header-only chunks from injecting blank lines into the merged CSV.
   • Ensured safe temporary directory management (exist_ok=True and cleanup in finally blocks).

4. Testing & Coverage:

   • Added comprehensive unit tests in fema_download_test.py covering:
     • Record count retrieval and error handling.
     • Direct bulk download success flow.
     • Direct bulk download failure with fallback to API pagination.
     • Multi-chunk pagination and header-only chunk merging without blank lines.

Verification

• Validation Report - link
• CRA Report - link
• Differ summary - link (All these deletions are expected and are from Source side changes)

[gemini-code-assist]

Code Review

This pull request introduces a direct bulk download option for the FEMA NFIP claims dataset as a faster and more reliable alternative to API pagination, which remains as a fallback. It also increases the pagination page size, improves error handling, and refines CSV chunk merging. The review feedback highlights two key improvements: raising an exception on bulk download failures to ensure proper logging instead of a silent fallback, and adding a check to prevent writing empty newlines when a downloaded chunk contains only a header.

[rohitkumarbhagat]

Is this page size supported?

[rohitkumarbhagat]

If total_records is 0 (e.g. if the upstream OpenFEMA metadata endpoint returns 0 due to a transient API issue or empty response), total_records == 0 evaluates to True indefinitely, causing an infinite pagination loop.

Consider removing total_records == 0 or from the while condition and adding an explicit pre-check to terminate immediately if total_records == 0:

if total_records == 0:
    logging.fatal("Total records returned 0 from API metadata. Cannot proceed.")
    raise RuntimeError("Total records returned 0.")

[rohitkumarbhagat]

Under app.run(main), absl.logging.fatal() aborts the process immediately with SIGABRT (exit code 134). Raising RuntimeError immediately afterwards or maintaining trailing control flow is redundant. Ensure fatal logging and exception handling are clean and consistent across standalone and library execution modes.

[rohitkumarbhagat]

Increasing PAGE_SIZE to 10000 is a great improvement (reducing API chunks from 2,650 to ~265). To further eliminate network overhead and prevent potential timeout/latency issues across ~265 chunks, consider streaming responses directly using a persistent requests.Session with connection pooling rather than making separate unpooled HEAD + GET requests via download_file on each chunk.

[rohitkumarbhagat]

output_dir = "input_file" (and default temp_dir = 'temp_fema_data') resolves relative to the current working directory (os.getcwd()). When executed from the repository root, this creates input_file/ in the repository root rather than inside statvar_imports/fema/flood_insurance_claims/.

Consider resolving paths relative to __file__:

script_dir = os.path.dirname(os.path.abspath(__file__))
output_dir = os.path.join(script_dir, "input_file")

[rohitkumarbhagat]

manifest.json processor flags:

The process.py command in manifest.json does not pass --existing_statvar_mcf or --output_counters.

Please add:

• --existing_statvar_mcf="gs://unresolved_mcf/scripts/statvar/stat_vars.mcf" (to reuse existing schema definitions and avoid duplicates)
• --output_counters="counters/counters.txt" (for pipeline counter telemetry in Cloud Batch)

[rohitkumarbhagat]

manifest.json source_files:

manifest.json currently only lists "source_files": ["input_file/fema_nfip_claims.csv"].

Please update "source_files" to include golden files, validation configs, and counters:

"source_files": [
  "input_file/fema_nfip_claims.csv",
  "golden_data/*.csv",
  "validation_config.json",
  "counters/*"
]

This ensures these files are copied to the version directory in GCS during batch execution for full historical reproducibility.

[rohitkumarbhagat]

validation_config.json date freshness check:

validation_config.json only configures DELETED_RECORDS_PERCENT and GOLDENS_CHECK. Because golden_summary_report.csv excludes MaxDate, there is no automated validation that data is being refreshed with recent dates.

Consider adding a scoped SQL_VALIDATOR rule in validation_config.json asserting that primary active aggregate StatVars (e.g. CountOfClaims_NaturalHazardInsurance_FEMAHighRiskFloodZone_BuildingStructureAndContents_FloodEvent) have a MaxDate within allowable lag (e.g. within 2–3 months of run date).

[rohitkumarbhagat]

Golden fixtures hygiene:

1. golden_observations.csv: Contains 2,349 rows, exceeding the guideline recommended 50–200 rows. Consider regenerating it filtered against gs://unresolved_mcf/import_validation/top_100k_places.csv.
2. golden_summary_report.csv: Contains 309 StatVars with hardcoded NumPlaces. As FEMA claims expand geographically over time, exact place count checks cause false positive golden check failures. Consider filtering against gs://unresolved_mcf/import_validation/nl_statvars.csv using validator_goldens.py.

[rohitkumarbhagat]

Verification on Page Size & Bulk Download Parity

1. Page Size Limit ($top=10000):

   • OpenFEMA API documentation confirms that $top defaults to 1,000 and supports a maximum of 10,000 records per request.
   • Live testing against https://www.fema.gov/api/open/v2/FimaNfipClaims?$format=csv&$top=10000&$skip=0 returns HTTP 200 with all 10,000 records in ~1.2s without throttling.

2. Bulk CSV Dataset Parity (bulk_url):

   • https://www.fema.gov/about/reports-and-data/openfema/FimaNfipClaims.csv contains the full historical dataset (~1.06 GB, ~2.72M records).
   • The CSV headers contain all 73 columns, matching the API CSV schema 1:1, and both are updated monthly (R/P1M) on OpenFEMA's release schedule.

3. Recommendation:

   • Direct bulk_url download is well-suited as the primary download method (~1–2 min download time vs 50+ hours).
   • PAGE_SIZE = 10000 ($top=10000) is safe and officially supported for API pagination fallback.