Skip to content

Optimize CDC air quality imports with sharding and scaled compute #2193

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
abhishekjaisw wants to merge 7 commits into
base: master
Choose a base branch
from
Open

Optimize CDC air quality imports with sharding and scaled compute #2193

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
7 commits
Select commit Hold shift + click to select a range
b37ef9a
Optimize CDC air quality imports with sharding and scaled compute
abhishekjaisw Aug 31, 2026
4ebc67e
Fix county sharding logic: initialize headers for all shards and prev…
abhishekjaisw Aug 31, 2026
5a7f855
Apply yapf Google code formatting to scripts/us_cdc/environmental_hea…
abhishekjaisw Aug 31, 2026
30e7938
Add automatic test artifact cleanup in parse_air_quality_test tearDown
abhishekjaisw Aug 31, 2026
4e0cd49
Remove redundant config_override block from manifest.json
abhishekjaisw Aug 31, 2026
1363a5f
Update README.md with sharded tMCF files and architecture documentation
abhishekjaisw Aug 31, 2026
2ac44d2
Merge branch 'master' into cdc_air_quality_sharding_and_scaling
abhishekjaisw Sep 1, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion scripts/us_cdc/environmental_health_toxicology/PM25CountyPollution.tmcf
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -10,7 +10,7 @@ variableMeasured: dcs:Mean_Concentration_AirPollutant_PM2.5
Node: E:PM25CountyPollution->E2
observationAbout: C:PM25CountyPollution->dcid
typeOf: dcs:StatVarObservation
observationDate: C:PPM25CountyPollution->date
observationDate: C:PM25CountyPollution->date
value: C:PM25CountyPollution->pm25_med_pred
observationPeriod: "P24H"
unit: MicrogramsPerCubicMeter
Expand Down
35 changes: 35 additions & 0 deletions scripts/us_cdc/environmental_health_toxicology/PM25CountyPollution_part1.tmcf
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,35 @@
Node: E:PM25CountyPollution->E1
observationAbout: C:PM25CountyPollution->dcid
typeOf: dcs:StatVarObservation
observationDate: C:PM25CountyPollution->date
value: C:PM25CountyPollution->pm25_mean_pred
observationPeriod: "P24H"
unit: MicrogramsPerCubicMeter
variableMeasured: dcs:Mean_Concentration_AirPollutant_PM2.5

Node: E:PM25CountyPollution->E2
observationAbout: C:PM25CountyPollution->dcid
typeOf: dcs:StatVarObservation
observationDate: C:PM25CountyPollution->date
value: C:PM25CountyPollution->pm25_med_pred
observationPeriod: "P24H"
unit: MicrogramsPerCubicMeter
variableMeasured: dcs:Median_Concentration_AirPollutant_PM2.5

Node: E:PM25CountyPollution->E3
observationAbout: C:PM25CountyPollution->dcid
typeOf: dcs:StatVarObservation
observationDate: C:PM25CountyPollution->date
value: C:PM25CountyPollution->pm25_max_pred
observationPeriod: "P24H"
unit: MicrogramsPerCubicMeter
variableMeasured: dcs:Max_Concentration_AirPollutant_PM2.5

Node: E:PM25CountyPollution->E4
observationAbout: C:PM25CountyPollution->dcid
typeOf: dcs:StatVarObservation
observationDate: C:PM25CountyPollution->date
value: C:PM25CountyPollution->pm25_pop_pred
observationPeriod: "P24H"
unit: MicrogramsPerCubicMeter
variableMeasured: dcs:PopulationWeighted_Concentration_AirPollutant_PM2.5
35 changes: 35 additions & 0 deletions scripts/us_cdc/environmental_health_toxicology/PM25CountyPollution_part2.tmcf
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,35 @@
Node: E:PM25CountyPollution->E1
observationAbout: C:PM25CountyPollution->dcid
typeOf: dcs:StatVarObservation
observationDate: C:PM25CountyPollution->date
value: C:PM25CountyPollution->pm25_mean_pred
observationPeriod: "P24H"
unit: MicrogramsPerCubicMeter
variableMeasured: dcs:Mean_Concentration_AirPollutant_PM2.5

Node: E:PM25CountyPollution->E2
observationAbout: C:PM25CountyPollution->dcid
typeOf: dcs:StatVarObservation
observationDate: C:PM25CountyPollution->date
value: C:PM25CountyPollution->pm25_med_pred
observationPeriod: "P24H"
unit: MicrogramsPerCubicMeter
variableMeasured: dcs:Median_Concentration_AirPollutant_PM2.5

Node: E:PM25CountyPollution->E3
observationAbout: C:PM25CountyPollution->dcid
typeOf: dcs:StatVarObservation
observationDate: C:PM25CountyPollution->date
value: C:PM25CountyPollution->pm25_max_pred
observationPeriod: "P24H"
unit: MicrogramsPerCubicMeter
variableMeasured: dcs:Max_Concentration_AirPollutant_PM2.5

Node: E:PM25CountyPollution->E4
observationAbout: C:PM25CountyPollution->dcid
typeOf: dcs:StatVarObservation
observationDate: C:PM25CountyPollution->date
value: C:PM25CountyPollution->pm25_pop_pred
observationPeriod: "P24H"
unit: MicrogramsPerCubicMeter
variableMeasured: dcs:PopulationWeighted_Concentration_AirPollutant_PM2.5
35 changes: 35 additions & 0 deletions scripts/us_cdc/environmental_health_toxicology/PM25CountyPollution_part3.tmcf
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,35 @@
Node: E:PM25CountyPollution->E1
observationAbout: C:PM25CountyPollution->dcid
typeOf: dcs:StatVarObservation
observationDate: C:PM25CountyPollution->date
value: C:PM25CountyPollution->pm25_mean_pred
observationPeriod: "P24H"
unit: MicrogramsPerCubicMeter
variableMeasured: dcs:Mean_Concentration_AirPollutant_PM2.5

Node: E:PM25CountyPollution->E2
observationAbout: C:PM25CountyPollution->dcid
typeOf: dcs:StatVarObservation
observationDate: C:PM25CountyPollution->date
value: C:PM25CountyPollution->pm25_med_pred
observationPeriod: "P24H"
unit: MicrogramsPerCubicMeter
variableMeasured: dcs:Median_Concentration_AirPollutant_PM2.5

Node: E:PM25CountyPollution->E3
observationAbout: C:PM25CountyPollution->dcid
typeOf: dcs:StatVarObservation
observationDate: C:PM25CountyPollution->date
value: C:PM25CountyPollution->pm25_max_pred
observationPeriod: "P24H"
unit: MicrogramsPerCubicMeter
variableMeasured: dcs:Max_Concentration_AirPollutant_PM2.5

Node: E:PM25CountyPollution->E4
observationAbout: C:PM25CountyPollution->dcid
typeOf: dcs:StatVarObservation
observationDate: C:PM25CountyPollution->date
value: C:PM25CountyPollution->pm25_pop_pred
observationPeriod: "P24H"
unit: MicrogramsPerCubicMeter
variableMeasured: dcs:PopulationWeighted_Concentration_AirPollutant_PM2.5
35 changes: 35 additions & 0 deletions scripts/us_cdc/environmental_health_toxicology/PM25CountyPollution_part4.tmcf
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,35 @@
Node: E:PM25CountyPollution->E1
observationAbout: C:PM25CountyPollution->dcid
typeOf: dcs:StatVarObservation
observationDate: C:PM25CountyPollution->date
value: C:PM25CountyPollution->pm25_mean_pred
observationPeriod: "P24H"
unit: MicrogramsPerCubicMeter
variableMeasured: dcs:Mean_Concentration_AirPollutant_PM2.5

Node: E:PM25CountyPollution->E2
observationAbout: C:PM25CountyPollution->dcid
typeOf: dcs:StatVarObservation
observationDate: C:PM25CountyPollution->date
value: C:PM25CountyPollution->pm25_med_pred
observationPeriod: "P24H"
unit: MicrogramsPerCubicMeter
variableMeasured: dcs:Median_Concentration_AirPollutant_PM2.5

Node: E:PM25CountyPollution->E3
observationAbout: C:PM25CountyPollution->dcid
typeOf: dcs:StatVarObservation
observationDate: C:PM25CountyPollution->date
value: C:PM25CountyPollution->pm25_max_pred
observationPeriod: "P24H"
unit: MicrogramsPerCubicMeter
variableMeasured: dcs:Max_Concentration_AirPollutant_PM2.5

Node: E:PM25CountyPollution->E4
observationAbout: C:PM25CountyPollution->dcid
typeOf: dcs:StatVarObservation
observationDate: C:PM25CountyPollution->date
value: C:PM25CountyPollution->pm25_pop_pred
observationPeriod: "P24H"
unit: MicrogramsPerCubicMeter
variableMeasured: dcs:PopulationWeighted_Concentration_AirPollutant_PM2.5
20 changes: 7 additions & 13 deletions scripts/us_cdc/environmental_health_toxicology/README.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -69,19 +69,13 @@ These data were collected as part of the [CDC National Environment Public Health
[`small_Palmer_expected.csv`](https://github.com/datacommonsorg/data/blob/master/scripts/us_cdc/environmental_health_toxicology/test_data/small_Palmer_expected.csv)

#### tMCFs
[`OzoneCensusTractPollution.tmcf`](https://github.com/datacommonsorg/data/blob/master/scripts/us_cdc/environmental_health_toxicology/OzoneCensusTractPollution.tmcf)

[`OzoneCountyPollution.tmcf`](https://github.com/datacommonsorg/data/blob/master/scripts/us_cdc/environmental_health_toxicology/OzoneCountyPollution.tmcf)

[`PalmerDroughtSeverityIndex.tmcf`](https://github.com/datacommonsorg/data/blob/master/scripts/us_cdc/environmental_health_toxicology/PalmerDroughtSeverityIndex.tmcf)

[`PM25CensusTractPollution.tmcf`](https://github.com/datacommonsorg/data/blob/master/scripts/us_cdc/environmental_health_toxicology/PM25CensusTractPollution.tmcf)

[`PM25CountyPollution.tmcf`](https://github.com/datacommonsorg/data/blob/master/scripts/us_cdc/environmental_health_toxicology/PM25CountyPollution.tmcf)

[`StandardizedPrecipitationEvapotranspirationIndex.tmcf`](https://github.com/datacommonsorg/data/blob/master/scripts/us_cdc/environmental_health_toxicology/StandardizedPrecipitationEvapotranspirationIndex.tmcf)

[`StandardizedPrecipitationIndex.tmcf`](https://github.com/datacommonsorg/data/blob/master/scripts/us_cdc/environmental_health_toxicology/StandardizedPrecipitationIndex.tmcf)
* [`OzoneCountyPollution.tmcf`](https://github.com/datacommonsorg/data/blob/master/scripts/us_cdc/environmental_health_toxicology/OzoneCountyPollution.tmcf)
* **Sharded PM2.5 County tMCFs:** [`PM25CountyPollution_part1.tmcf`](https://github.com/datacommonsorg/data/blob/master/scripts/us_cdc/environmental_health_toxicology/PM25CountyPollution_part1.tmcf), [`part2`](https://github.com/datacommonsorg/data/blob/master/scripts/us_cdc/environmental_health_toxicology/PM25CountyPollution_part2.tmcf), [`part3`](https://github.com/datacommonsorg/data/blob/master/scripts/us_cdc/environmental_health_toxicology/PM25CountyPollution_part3.tmcf), [`part4`](https://github.com/datacommonsorg/data/blob/master/scripts/us_cdc/environmental_health_toxicology/PM25CountyPollution_part4.tmcf)
* **Sharded PM2.5 Census Tract tMCFs:** [`PM25CensusTractPollution_part1.tmcf`](https://github.com/datacommonsorg/data/blob/master/scripts/us_cdc/environmental_health_toxicology/PM25CensusTractPollution_part1.tmcf), [`part2`](https://github.com/datacommonsorg/data/blob/master/scripts/us_cdc/environmental_health_toxicology/PM25CensusTractPollution_part2.tmcf), [`part3`](https://github.com/datacommonsorg/data/blob/master/scripts/us_cdc/environmental_health_toxicology/PM25CensusTractPollution_part3.tmcf), [`part4`](https://github.com/datacommonsorg/data/blob/master/scripts/us_cdc/environmental_health_toxicology/PM25CensusTractPollution_part4.tmcf)
* **Sharded Ozone Census Tract tMCFs:** [`OzoneCensusTractPollution_part1.tmcf`](https://github.com/datacommonsorg/data/blob/master/scripts/us_cdc/environmental_health_toxicology/OzoneCensusTractPollution_part1.tmcf), [`part2`](https://github.com/datacommonsorg/data/blob/master/scripts/us_cdc/environmental_health_toxicology/OzoneCensusTractPollution_part2.tmcf), [`part3`](https://github.com/datacommonsorg/data/blob/master/scripts/us_cdc/environmental_health_toxicology/OzoneCensusTractPollution_part3.tmcf), [`part4`](https://github.com/datacommonsorg/data/blob/master/scripts/us_cdc/environmental_health_toxicology/OzoneCensusTractPollution_part4.tmcf)
* [`PalmerDroughtSeverityIndex.tmcf`](https://github.com/datacommonsorg/data/blob/master/scripts/us_cdc/environmental_health_toxicology/PalmerDroughtSeverityIndex.tmcf)
* [`StandardizedPrecipitationEvapotranspirationIndex.tmcf`](https://github.com/datacommonsorg/data/blob/master/scripts/us_cdc/environmental_health_toxicology/StandardizedPrecipitationEvapotranspirationIndex.tmcf)
* [`StandardizedPrecipitationIndex.tmcf`](https://github.com/datacommonsorg/data/blob/master/scripts/us_cdc/environmental_health_toxicology/StandardizedPrecipitationIndex.tmcf)

### Import Procedure

Expand Down
21 changes: 7 additions & 14 deletions scripts/us_cdc/environmental_health_toxicology/download_files.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -36,20 +36,13 @@ def download_files(importname, configs):
@retry(tries=3, delay=2, backoff=2)
def download_with_retry(url, input_file_name):
logging.info(f"Downloading file from URL: {url}")
response = requests.get(url)
response.raise_for_status()
if response.status_code == 200:
if not response.content:
logging.fatal(
f"No data available for URL: {url}. Aborting download.")
return
filename = os.path.join(_INPUT_FILE_PATH, input_file_name)
with file_util.FileIO(filename, 'wb') as f:
f.write(response.content)
else:
logging.error(
f"Failed to download file from URL: {url}. Status code: {response.status_code}"
)
filename = os.path.join(_INPUT_FILE_PATH, input_file_name)
with requests.get(url, stream=True) as response:
response.raise_for_status()
with open(filename, 'wb') as f:
for chunk in response.iter_content(chunk_size=16 * 1024 * 1024):
if chunk:
f.write(chunk)

try:
for config in configs:
Expand Down
56 changes: 36 additions & 20 deletions scripts/us_cdc/environmental_health_toxicology/manifest.json
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -12,13 +12,14 @@
"parse_air_quality.py CDC_PM25CensusTract"
],
"source_files": [
"input/*.gz"
"input_files/*"
],
"resource_limits": {
"cpu": 8,
"memory": 64,
"disk": 100
"cpu": 32,
"memory": 512,
"disk": 2000
},
"validation_config_file": "validation_config.json",
"import_inputs": [
{
"template_mcf": "PM25CensusTractPollution_part1.tmcf",
Expand Down Expand Up @@ -51,13 +52,14 @@
"parse_air_quality.py CDC_OzoneCensusTract"
],
"source_files": [
"input/*.gz"
"input_files/*"
],
"resource_limits": {
"cpu": 8,
"memory": 64,
"disk": 100
"cpu": 32,
"memory": 512,
"disk": 2000
},
"validation_config_file": "validation_config.json",
"import_inputs": [
{
"template_mcf": "OzoneCensusTractPollution_part1.tmcf",
Expand Down Expand Up @@ -92,17 +94,30 @@
"source_files": [
"input_files/*"
],
"resource_limits": {
"cpu": 32,
"memory": 512,
"disk": 500
},
"validation_config_file": "validation_config.json",
"import_inputs": [
{
"template_mcf": "PM25CountyPollution.tmcf",
"cleaned_csv": "output/PM25county.csv"
"template_mcf": "PM25CountyPollution_part1.tmcf",
"cleaned_csv": "output/PM25county_0.csv"
},
{
"template_mcf": "PM25CountyPollution_part2.tmcf",
"cleaned_csv": "output/PM25county_1.csv"
},
{
"template_mcf": "PM25CountyPollution_part3.tmcf",
"cleaned_csv": "output/PM25county_2.csv"
},
{
"template_mcf": "PM25CountyPollution_part4.tmcf",
"cleaned_csv": "output/PM25county_3.csv"
}
],
"resource_limits": {
"cpu": 8,
"memory": 128,
"disk": 200
},
"cron_schedule": "0 1 4 * *"
},
{
Expand All @@ -119,17 +134,18 @@
"source_files": [
"input_files/*"
],
"resource_limits": {
"cpu": 32,
"memory": 512,
"disk": 500
},
"validation_config_file": "validation_config.json",
"import_inputs": [
{
"template_mcf": "OzoneCountyPollution.tmcf",
"cleaned_csv": "output/OzoneCounty.csv"
}
],
"resource_limits": {
"cpu": 16,
"memory": 512,
"disk": 500
},
"cron_schedule": "0 1 5 * *"
}
]
Expand Down
Loading
Loading