CDCWonderNatality : Automation - #2233
kartik-s21 wants to merge 62 commits into
Conversation
Code fix unenergy
…nd pipeline scripts - Implement automated download script download.sh from GCS repository - Implement unified process.py pipeline for country, state, and county tables - Add unit tests in process_test.py and ensure deterministic sorting in preprocess.py - Add manifest.json for Cloud Run/Batch automation (annual cadence, 3 import inputs, no node_mcf) - Add validation_config.json with empty import, lint error, and missing refs checks - Update README.md with comprehensive documentation and usage instructions
|
Warning Gemini encountered an error creating the review. You can try again by commenting |
|
/gemini review |
![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 an automated end-to-end data processing pipeline for CDC Wonder Natality data, including new scripts for downloading, processing, and testing, along with a manifest and validation configuration. The review feedback highlights several improvement opportunities: making the geographic classification logic in process.py more robust, casting aggregated count statistics to integers (Int64) to prevent float representation, correcting a fallback test data copy that mixes state and county resolutions, ensuring output verification checks for non-empty files, and replacing the file command in download.sh with a more portable unzip -t check.
| sample_geo = (sample_df['Geo'].dropna().iloc[0] if | ||
| not sample_df['Geo'].dropna().empty else '') | ||
| # State DCIDs are geoId/XX (length 9), County DCIDs are geoId/XXXXX (length 12) | ||
| if len(sample_geo) <= 9: | ||
| state_csvs.append(f) | ||
| else: | ||
| county_csvs.append(f) |
There was a problem hiding this comment.
The current classification logic relies on len(sample_geo) <= 9 to distinguish between state and county DCIDs. However, this is fragile: an empty sample_geo (length 0) or a country DCID like country/USA (length 11) could be misclassified. Since US state DCIDs are always geoId/XX (length 8) and county DCIDs are geoId/XXXXX (length 11), we can make this check much more robust by explicitly matching these patterns.
| sample_geo = (sample_df['Geo'].dropna().iloc[0] if | |
| not sample_df['Geo'].dropna().empty else '') | |
| # State DCIDs are geoId/XX (length 9), County DCIDs are geoId/XXXXX (length 12) | |
| if len(sample_geo) <= 9: | |
| state_csvs.append(f) | |
| else: | |
| county_csvs.append(f) | |
| sample_geo = (sample_df['Geo'].dropna().iloc[0] if | |
| not sample_df['Geo'].dropna().empty else '') | |
| # State DCIDs are geoId/XX (length 8), County DCIDs are geoId/XXXXX (length 11) | |
| if sample_geo.startswith('country/'): | |
| country_csvs.append(f) | |
| elif sample_geo.startswith('geoId/') and len(sample_geo) == 11: | |
| county_csvs.append(f) | |
| elif sample_geo.startswith('geoId/') and len(sample_geo) == 8: | |
| state_csvs.append(f) | |
| else: | |
| logging.warning(f'Could not determine geo level for {f} with geo {sample_geo}') |
References
- A dcid is a unique identifier that maps to a single, consistent set of properties. Ensuring correct identification of DCIDs (like state vs county) prevents mapping issues.
| count_df['Quantity'] = pd.to_numeric(count_df['Quantity'], | ||
| errors='coerce') | ||
| country_df = count_df.groupby(['Year', 'StatVar'], | ||
| as_index=False)['Quantity'].sum() | ||
| country_df.sort_values(by=['Year', 'StatVar'], inplace=True) | ||
| country_df.to_csv(country_out, index=False) |
There was a problem hiding this comment.
When converting Quantity to numeric and performing a .sum(), pandas will represent the resulting values as floats (e.g., 3613647.0). For count statistics, we should ensure they are formatted as integers in the output CSV to prevent schema/type mismatch issues during ingestion. Casting the column to the nullable integer type Int64 resolves this.
| count_df['Quantity'] = pd.to_numeric(count_df['Quantity'], | |
| errors='coerce') | |
| country_df = count_df.groupby(['Year', 'StatVar'], | |
| as_index=False)['Quantity'].sum() | |
| country_df.sort_values(by=['Year', 'StatVar'], inplace=True) | |
| country_df.to_csv(country_out, index=False) | |
| count_df['Quantity'] = pd.to_numeric(count_df['Quantity'], | |
| errors='coerce') | |
| country_df = count_df.groupby(['Year', 'StatVar'], | |
| as_index=False)['Quantity'].sum() | |
| country_df['Quantity'] = country_df['Quantity'].astype('Int64') | |
| country_df.sort_values(by=['Year', 'StatVar'], inplace=True) | |
| country_df.to_csv(country_out, index=False) |
| county_csv = os.path.join(output_path, 'county.csv') | ||
| if not os.path.exists(county_csv) and os.path.exists(tmp_state_csv): | ||
| shutil.copyfile(tmp_state_csv, county_csv) |
There was a problem hiding this comment.
Copying state-level data (tmp_state_csv) directly to county_csv is incorrect because they represent different geographic resolutions (state DCIDs like geoId/01 vs county DCIDs like geoId/01003). This will result in invalid county-level data if the pipeline is run with --use_test_data. Consider running the preprocessing script with the county config if county test data is available, or at least log a warning/error instead of silently copying mismatched geographic data.
| f for f in required_outputs | ||
| if not os.path.exists(os.path.join(output_path, f)) | ||
| ] | ||
| if missing: |
There was a problem hiding this comment.
The comment above states that the pipeline verifies that all required output files exist and are non-empty. However, the code only checks os.path.exists. To align with the comment and prevent downstream ingestion of empty files, we should also verify that the file size is greater than 0.
missing = [
f for f in required_outputs
if not os.path.exists(os.path.join(output_path, f)) or os.path.getsize(os.path.join(output_path, f)) == 0
]| if [ -f "$SCRIPT_DIR/source_data/county.zip" ]; then | ||
| if file "$SCRIPT_DIR/source_data/county.zip" | grep -q "Zip archive"; then | ||
| echo "Extracting raw county source data..." | ||
| unzip -q -o "$SCRIPT_DIR/source_data/county.zip" -d "$INPUT_DIR/" || true | ||
| fi | ||
| fi |
There was a problem hiding this comment.
The file command is often not installed in minimal execution environments (such as standard Docker containers used in CI/CD or Cloud Batch). Using unzip -t is a much more portable and standard way to verify if a file is a valid zip archive.
| if [ -f "$SCRIPT_DIR/source_data/county.zip" ]; then | |
| if file "$SCRIPT_DIR/source_data/county.zip" | grep -q "Zip archive"; then | |
| echo "Extracting raw county source data..." | |
| unzip -q -o "$SCRIPT_DIR/source_data/county.zip" -d "$INPUT_DIR/" || true | |
| fi | |
| fi | |
| if [ -f "$SCRIPT_DIR/source_data/county.zip" ]; then | |
| if unzip -t "$SCRIPT_DIR/source_data/county.zip" &>/dev/null; then | |
| echo "Extracting raw county source data..." | |
| unzip -q -o "$SCRIPT_DIR/source_data/county.zip" -d "$INPUT_DIR/" || true | |
| fi | |
| fi |
There was a problem hiding this comment.
Code Review
This pull request introduces automation scripts, configurations, and tests for the CDC Wonder Natality import, including an end-to-end processing pipeline (process.py) and a download script (download.sh). Feedback on the new pipeline highlights critical issues in process.py: the geographic classification logic for sample_geo contains flaws regarding DCID lengths and empty values; converting the Quantity column to numeric can result in float representations instead of integers; and copying state-level data to county-level output as a fallback introduces mismatched geographic associations.
| sample_geo = (sample_df['Geo'].dropna().iloc[0] if | ||
| not sample_df['Geo'].dropna().empty else '') | ||
| # State DCIDs are geoId/XX (length 9), County DCIDs are geoId/XXXXX (length 12) | ||
| if len(sample_geo) <= 9: | ||
| state_csvs.append(f) | ||
| else: | ||
| county_csvs.append(f) |
There was a problem hiding this comment.
The classification logic based on the length of sample_geo has a few issues:
- Incorrect Lengths in Comment: State DCIDs like
geoId/01have a length of 8 (not 9), and County DCIDs likegeoId/01001have a length of 11 (not 12). - Empty Geo Handling: If
sample_df['Geo']is entirely empty/NaN,sample_geobecomes''(length 0), which is<= 9and will be incorrectly classified as a state CSV. - Country DCID Misclassification: If
sample_geoiscountry/USA(length 11), it is> 9and will be incorrectly classified as a county CSV.
Consider a more robust classification approach, such as checking if the DCID starts with geoId/ and then checking the length of the numeric part, or checking for country explicitly.
| sample_geo = (sample_df['Geo'].dropna().iloc[0] if | |
| not sample_df['Geo'].dropna().empty else '') | |
| # State DCIDs are geoId/XX (length 9), County DCIDs are geoId/XXXXX (length 12) | |
| if len(sample_geo) <= 9: | |
| state_csvs.append(f) | |
| else: | |
| county_csvs.append(f) | |
| sample_geo = (sample_df['Geo'].dropna().iloc[0] if | |
| not sample_df['Geo'].dropna().empty else '') | |
| if not sample_geo: | |
| logging.warning(f'Could not classify {f}: Geo column is empty') | |
| elif 'country' in sample_geo or 'USA' in sample_geo: | |
| country_csvs.append(f) | |
| elif sample_geo.startswith('geoId/'): | |
| parts = sample_geo.split('/') | |
| if len(parts) > 1 and len(parts[1]) == 2: | |
| state_csvs.append(f) | |
| else: | |
| county_csvs.append(f) | |
| else: | |
| logging.warning(f'Could not classify {f} with Geo: {sample_geo}') |
| count_df['Quantity'] = pd.to_numeric(count_df['Quantity'], | ||
| errors='coerce') |
There was a problem hiding this comment.
Converting Quantity to numeric without specifying an integer type can result in float representation (e.g., 3613647.0) in the generated country.csv if there are any missing values or due to pandas default upcasting. Since these are birth counts, they should be represented as integers.
Using pandas' nullable integer type Int64 (capital 'I') will preserve integer formatting while safely handling any potential NaN values.
| count_df['Quantity'] = pd.to_numeric(count_df['Quantity'], | |
| errors='coerce') | |
| count_df['Quantity'] = pd.to_numeric(count_df['Quantity'], | |
| errors='coerce').astype('Int64') |
| county_csv = os.path.join(output_path, 'county.csv') | ||
| if not os.path.exists(county_csv) and os.path.exists(tmp_state_csv): | ||
| shutil.copyfile(tmp_state_csv, county_csv) |
There was a problem hiding this comment.
Copying state.csv to county.csv as a fallback is conceptually incorrect because state-level data contains state DCIDs (e.g., geoId/01), whereas county-level data is expected to contain county DCIDs (e.g., geoId/01003). Processing state-level data with the county TMCF mapping (county/output.tmcf) will result in invalid or mismatched geographic associations in Data Commons.
If county-level test data is not available or cannot be generated, it is better to skip generating county.csv in the fallback path or raise a clear warning/error, rather than copying mismatched geographic data.
- Use distinct output TMCF paths (country.tmcf, state.tmcf, county.tmcf) in manifest.json to prevent Cloud Batch GCS basename collision. - Remove dangling validation_config_file from manifest.json and README.md. - Ensure chronological sorting and filename-based routing in process.py. - Use regex Geo validation (geoId/XX vs geoId/XXXXX) and cast aggregated quantities to Int64. - Tighten error handling in download.sh: require gcloud, remove silent failure suppression, and verify all geographic resolution levels. - Track Buganizer issue b/555054352.
1 participant
Summary
This PR automates the ingestion pipeline for
CDCWonderNatality, migrating the import to the automated Cloud Batch framework and aligning with Google3 CL 986267877.Bug & CL References
Changes
manifest.json):import_inputsacross Country, State, and County levels pointing to distinct template paths (output/country.tmcf,output/state.tmcf,output/county.tmcf) to prevent GCS basename overwrite collisions.0 0 1 6 *.process.py):_extract_start_year) across reporting revisions (95-02,03-06,07-20,16-22).os.path.basename(f).lower()for safe file classification.Geocolumn (^geoId/\d{2}$for state vs^geoId/\d{5}$for county).Quantityto nullable integerInt64.output/.download.sh):gs://unresolved_mcf/cdc/wonder/natality/).process_test.pyandpreprocess_test.py.Test Execution & Validation Artifacts
cdcwondernatality-samnotra-20260927-035321gs://datcom-import-test/scripts/us_cdc/natality/CDCWonderNatality/2026_09_26T20_56_16_779734_07_00/