diff --git a/scripts/us_census/acs5yr/subject_tables/S2303/README.md b/scripts/us_census/acs5yr/subject_tables/S2303/README.md index a0cd358eda..ac4958e8f3 100644 --- a/scripts/us_census/acs5yr/subject_tables/S2303/README.md +++ b/scripts/us_census/acs5yr/subject_tables/S2303/README.md @@ -1,12 +1,82 @@ -# US census ACS5YR S2303 Subject Table import +# US Census ACS5YR S2303 Subject Table Import This import gives the count of people in the age range 16-64 years who have worked 1 or more weeks in the past 12 months. It provides a further breakdown of the data on the gender, number of working weeks and the amount of hours worked per week. -Years: 2010-2019 -Geo : Country, State, County and Place +- **Years**: 2010-2024 +- **Geo**: Country, State, County, County Subdivision, Census Tract, Place/City, Congressional District, ZCTA, School District, CBSA, Census Division -Important Notes : +## Important Notes 1. Median age of workers is not available from 2010-2014. 2. Data for "Workers 16 to 64 years who worked full-time, year-round" is not available from 2010-2014. -3. The breakdown of usual hours worked from 2010 - 2014 is a subset of the ones available from 2015 - 2019. -4. The values are published as counts from 2015 - 2019 and as percentages from 2010 - 2014. +3. The breakdown of usual hours worked from 2010 - 2014 is a subset of the ones available from 2015 - 2024. +4. The values are published as counts from 2015 - 2024 and as percentages from 2010 - 2014. + +## Download Step + +Run `census_api_data_downloader.py` from `scripts/us_census/api_utils` for each 3-year range using `--start_year` / `--end_year` and copy each generated `S2303.zip` into `input_files/`: + +```bash +cd scripts/us_census/api_utils +mkdir -p ../acs5yr/subject_tables/S2303/input_files + +for range in 2010_2012 2013_2015 2016_2018 2019_2021 2022_2024; do + start_yr=${range%_*} + end_yr=${range#*_} + python3 census_api_data_downloader.py \ + --dataset=acs/acs5/subject \ + --table_id=S2303 \ + --start_year=$start_yr \ + --end_year=$end_yr \ + --summary_levels=010,030,040,050,060,140,160,310,500,860,950,960,970 \ + --output_path=~/us_census \ + --api_key= + cp ~/us_census/acs/acs5/subject/S2303/S2303.zip \ + ../acs5yr/subject_tables/S2303/input_files/s2303_${range}.zip +done +``` + +## Process Step + +Run `process.py` from `scripts/us_census/acs5yr/subject_tables/S2303`: + +```bash +cd ../acs5yr/subject_tables/S2303 + +for b in s2303_2010_2012 s2303_2013_2015 s2303_2016_2018 s2303_2019_2021 s2303_2022_2024; do + python3 process.py \ + --option=all \ + --table_prefix=$b \ + --has_percent=True \ + --debug=False \ + --spec_path=./S2303_spec.json \ + --input_path=input_files/${b}.zip \ + --output_dir=./ + mv ${b}_output.tmcf s2303_output.tmcf +done +``` + +- `--option=all`: Runs both column map generation (`_column_map.json`) and observation processing (`_cleaned.csv`, `_output.mcf`, `_output.tmcf`, `_summary.json`). +- `--has_percent=True`: Converts percentage estimates (`2010`–`2014`) into counts using the mapped denominator keys in `S2303_spec.json` (`S2303SubjectTableDataLoader` in `process.py` restricts conversion to `percent_years = ['2014', '2013', '2012', '2011', '2010']`). +- Only a single shared `s2303_output.tmcf` needs to be retained in the final import folder alongside the 5 `*_output.mcf` and 5 `*_cleaned.csv` batch files. + +## Batch Splitting & Runtime Optimization + +Processing all 15 years (`2010`–`2024`, ~80 GiB intermediate CSV / ~459M rows) as a single `.zip` requires ~155 GiB RAM during the final `pd.read_csv()` cleanup step and spills to swap. Splitting into **5 3-year `.zip` batches** keeps each run in RAM with zero swap thrashing: + +| Batch / `--table_prefix` | CSV Files Inside `.zip` | Output Files Generated | +| :--- | :--- | :--- | +| `s2303_2010_2012` | `ACSST5Y2010*.csv`, `ACSST5Y2011*.csv`, `ACSST5Y2012*.csv` | `s2303_2010_2012_{cleaned.csv, output.mcf, summary.json}` | +| `s2303_2013_2015` | `ACSST5Y2013*.csv`, `ACSST5Y2014*.csv`, `ACSST5Y2015*.csv` | `s2303_2013_2015_{cleaned.csv, output.mcf, summary.json}` | +| `s2303_2016_2018` | `ACSST5Y2016*.csv`, `ACSST5Y2017*.csv`, `ACSST5Y2018*.csv` | `s2303_2016_2018_{cleaned.csv, output.mcf, summary.json}` | +| `s2303_2019_2021` | `ACSST5Y2019*.csv`, `ACSST5Y2020*.csv`, `ACSST5Y2021*.csv` | `s2303_2019_2021_{cleaned.csv, output.mcf, summary.json}` | +| `s2303_2022_2024` | `ACSST5Y2022*.csv`, `ACSST5Y2023*.csv`, `ACSST5Y2024*.csv` | `s2303_2022_2024_{cleaned.csv, output.mcf, summary.json}` | + +### Runtime Comparison +| Method | Total Wall-Clock Time | Time Saved | Notes | +| :--- | :---: | :---: | :--- | +| **1 Single `.zip` (`2010`–`2024`)** | ~110 – 120 mins | — | Spills to swap during final `pd.read_csv()` rewrite. | +| **5 Split `.zip`s (Sequential)** | ~45 – 55 mins | ~60 – 65 mins (~2x faster) | Each batch (~16 GiB CSV) fits in RAM with zero swap. | +| **5 Split `.zip`s (Parallel, 2–3 jobs)** | ~18 – 22 mins | ~90 – 100 mins (~5x faster) | Utilizes multiple cores while staying under system RAM. | + + + diff --git a/scripts/us_census/acs5yr/subject_tables/S2303/process.py b/scripts/us_census/acs5yr/subject_tables/S2303/process.py index 963ad6c9a3..27064bef09 100644 --- a/scripts/us_census/acs5yr/subject_tables/S2303/process.py +++ b/scripts/us_census/acs5yr/subject_tables/S2303/process.py @@ -1,149 +1,253 @@ -"""Process data from Census ACS5Year Subject Table S2303.""" +# Copyright 2021 Google LLC +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +"""Process data from Census ACS 5-Year Subject Table S2303.""" + +import json import os import sys -import pandas as pd -import numpy as np -# pylint: disable=missing-module-docstring -# pylint: disable=missing-class-docstring -# pylint: disable=missing-function-docstring +from absl import app +from absl import flags +from absl import logging +import numpy as np +import pandas as pd -#pylint: disable=wrong-import-position -#pylint: disable=import-error +# pylint: disable=wrong-import-position +# pylint: disable=import-error _CODEDIR = os.path.dirname(os.path.realpath(__file__)) sys.path.insert(1, os.path.join(_CODEDIR, '..', 'common')) -from data_loader import SubjectTableDataLoaderBase # commit hash - 4bc5102 +from data_loader import SubjectTableDataLoaderBase +from generate_col_map import process_zip_file from resolve_geo_id import convert_to_place_dcid -#pylint: enable=wrong-import-position -#pylint: enable=import-error +# pylint: enable=wrong-import-position +# pylint: enable=import-error + +FLAGS = flags.FLAGS class S2303SubjectTableDataLoader(SubjectTableDataLoaderBase): - """ - A child class for the S2303 import. - """ - - def _convert_percent_to_numbers(self, df): - df_cols = df.columns.tolist() - for count_col, percent_cols in self.json_spec['denominators'].items(): - if count_col in df_cols: - for col in percent_cols: - if col in df_cols: - df[col] = df[col].astype(float) - df[count_col] = df[count_col].astype(float) - df[col] = (df[col] / 100) * df[count_col] - df[col] = df[col].round(3) - return df + """Data loader for Census ACS 5-Year Subject Table S2303.""" def _process_dataframe(self, df, filename): - """processes a dataframe read from a csv file""" + """Processes a dataframe read from a CSV file.""" df = self._replace_ignore_values_with_nan(df) year = filename.split(f'ACSST{self.estimate_period}Y')[1][:4] - print(f"Processing: {filename}", end=" | ", flush=True) - - # Process only at a country, state, county and place level - df = df[~((df['id'].str.startswith('950')) | - (df['id'].str.startswith('960')) | - (df['id'].str.startswith('970')))] + logging.info(f'Processing: {filename}') - # Only calculating percentages for selected years - percent_years = ['2014', '2013', '2012', '2011', '2010'] + # Only convert percentages to counts for 2010-2014 (2015+ report counts directly). + percent_years = self.json_spec.get( + 'percent_years', ['2014', '2013', '2012', '2011', '2010']) if self.has_percent and year in percent_years: - print("Converting percent to numbers", end=" | ", flush=True) + logging.info(f'Converting percent to numbers for {year}') df = self._convert_percent_to_numbers(df) - # if column map is not available generate, will rarely be False column_map = self.column_map[year] - # add stats to dict_counter for current year + # Add stats to counter_dict for current year. self.counter_dict[year] = { - "filename": + 'filename': filename, - "number of columns in dataset": + 'number of columns in dataset': df.shape[1], - "number of rows in dataset": + 'number of rows in dataset': df.shape[0], - "number of statVars generated for columns": + 'number of statVars generated for columns': len(list(column_map.keys())), - "number of observations": + 'number of unique geos': 0, - "number of unique StatVars with observations": + 'number of observations': + 0, + 'number of unique StatVars with observations': 0, } - csv_file = open(self.clean_csv_path, 'a') - place_geo_ids = df['id'].apply(convert_to_place_dcid) - - # update the clean csv - for column in df.columns.tolist(): - if column in column_map: - obs_df = pd.DataFrame(columns=self.csv_columns) - obs_df['Place'] = place_geo_ids - obs_df['StatVar'] = column_map[column]['Node'] - obs_df['Quantity'] = df[column].values.tolist() - # add unit to the csv - try: - unit = column_map[column]['unit'] - del column_map[column]['unit'] - except KeyError: - unit = np.nan - obs_df['Unit'] = unit - - # add scaling factor to the csv - try: - scaling_factor = column_map[column]['scalingFactor'] - del column_map[column]['scalingFactor'] - except KeyError: - unit = np.nan - scaling_factor = np.nan - - # if StatVar not in mcf_dict, add dcid - dcid = column_map[column]['Node'] - - if dcid not in self.mcf_dict: - ## key --> node dcid - self.mcf_dict[dcid] = {} - ## add pvs to dict - for key, value in column_map[column].items(): - if key != 'Node': - self.mcf_dict[dcid][key] = value - - obs_df['ScalingFactor'] = scaling_factor - obs_df['Year'] = year - obs_df['Column'] = column - - # Replace empty places (unresolved geoIds) as null values - obs_df['Place'].replace('', np.nan, inplace=True) - - # Drop rows with observations for empty (null) values - obs_df.dropna(subset=['Place', 'Quantity'], - axis=0, - inplace=True) - - # Write the processed observations to the clean_csv - if self.year_count == 0: - obs_df.to_csv(csv_file, header=True, index=False, mode='w') - else: - obs_df.to_csv(csv_file, header=False, index=False, mode='a') - self.year_count += 1 - - # update stats for the year: - self.counter_dict[year]["number of unique geos"] = len( - obs_df['Place'].unique()) - self.counter_dict[year][ - "number of observations"] += obs_df.shape[0] - self.counter_dict[year][ - "number of unique StatVars with observations"] += len( - obs_df['StatVar'].unique()) - self.counter_dict[year]["number of StatVars in mcf_dict"] = len( - list(self.mcf_dict.keys())) - csv_file.close() + if 'id' in df.columns: + place_geo_ids = df['id'].apply(convert_to_place_dcid) + else: + place_geo_ids = df['Geography'].apply(convert_to_place_dcid) + + unique_geos = set() + unique_statvars = set() + + with open(self.clean_csv_path, 'a') as csv_file: + for column in df.columns.tolist(): + if column in column_map: + obs_df = pd.DataFrame(columns=self.csv_columns) + obs_df['Place'] = place_geo_ids + obs_df['StatVar'] = column_map[column]['Node'] + obs_df['Quantity'] = df[column].values.tolist() + + obs_df['Unit'] = column_map[column].get('unit', np.nan) + obs_df['ScalingFactor'] = column_map[column].get( + 'scalingFactor', np.nan) + + dcid = column_map[column]['Node'] + if dcid not in self.mcf_dict: + self.mcf_dict[dcid] = {} + for key, value in column_map[column].items(): + if key not in ('Node', 'unit', 'scalingFactor'): + self.mcf_dict[dcid][key] = value + + obs_df['Year'] = year + obs_df['Column'] = column + + # Replace empty places (unresolved geoIds) as null values. + obs_df['Place'] = obs_df['Place'].replace('', np.nan) + + # Drop rows with observations for empty (null) values. + obs_df.dropna(subset=['Place', 'Quantity'], + axis=0, + inplace=True) + + obs_df.to_csv(csv_file, + header=(self.year_count == 0), + index=False) + self.year_count += 1 + + unique_geos.update(obs_df['Place'].unique()) + unique_statvars.update(obs_df['StatVar'].unique()) + self.counter_dict[year]['number of unique geos'] = len( + unique_geos) + self.counter_dict[year][ + 'number of observations'] += obs_df.shape[0] + self.counter_dict[year][ + 'number of unique StatVars with observations'] = len( + unique_statvars) + self.counter_dict[year][ + 'number of StatVars in mcf_dict'] = len(self.mcf_dict) + n_obs = self.counter_dict[year]['number of observations'] n_sv = self.counter_dict[year][ 'number of unique StatVars with observations'] n_geos = self.counter_dict[year]['number of unique geos'] - print( - f" Completed with {n_obs} observation for {n_sv} StatVars at " - f"{n_geos} places. ", - flush=True) + logging.info( + f'Completed with {n_obs} observations for {n_sv} StatVars at ' + f'{n_geos} places.') + + +def process_subject_tables(table_prefix, + input_path, + column_map_path, + spec_path, + output_dir='./', + has_percent=False, + debug=False, + estimate_period=5, + delimiter='!!', + decimal_places=3, + header_row=1): + """Processes S2303 zip/directory/csv inputs into cleaned CSV, MCF, and TMCF.""" + data_loader = S2303SubjectTableDataLoader(table_id=table_prefix, + col_delimiter=delimiter, + has_percent=has_percent, + debug=debug, + output_path_dir=output_dir, + json_spec=spec_path, + column_map_path=column_map_path, + decimal_places=decimal_places, + estimate_period=estimate_period, + header_row=header_row) + + if os.path.isdir(input_path): + data_loader._process_dir(input_path) + else: + _, file_extension = os.path.splitext(input_path) + if file_extension == '.zip' and os.path.isfile(input_path): + data_loader._process_zip_file(input_path) + elif file_extension == '.csv' and os.path.isfile(input_path): + data_loader._process_csv_file(input_path) + else: + logging.fatal( + f'Unsupported or non-existent input_path: {input_path}') + + +def set_column_map(input_path, spec_path, output_dir, table_prefix=None): + os.makedirs(output_dir, exist_ok=True) + generated_col_map = process_zip_file(input_path, + spec_path, + write_output=False) + col_map_filename = (f'{table_prefix}_column_map.json' + if table_prefix else 'column_map.json') + col_map_path = os.path.join(output_dir, col_map_filename) + with open(col_map_path, 'w') as f: + json.dump(generated_col_map, f, indent=4) + return col_map_path + + +def main(argv): + del argv + option = FLAGS.option.lower() + if option not in ('colmap', 'process', 'all'): + logging.fatal( + f"Invalid --option '{FLAGS.option}'. Must be one of: 'colmap', " + "'process', 'all'.") + table_prefix = FLAGS.table_prefix + spec_path = FLAGS.spec_path + input_path = FLAGS.input_path + output_dir = FLAGS.output_dir + has_percent = FLAGS.has_percent + debug = FLAGS.debug + + _, file_extension = os.path.splitext(input_path) + + if file_extension == '.zip': + col_map_path = os.path.join(output_dir, + f'{table_prefix}_column_map.json') + if option in ('colmap', 'all'): + col_map_path = set_column_map(input_path, + spec_path, + output_dir, + table_prefix=table_prefix) + if option in ('process', 'all'): + if not os.path.exists(col_map_path): + fallback_path = os.path.join(output_dir, 'column_map.json') + if os.path.exists(fallback_path): + col_map_path = fallback_path + process_subject_tables(table_prefix=table_prefix, + input_path=input_path, + output_dir=output_dir, + column_map_path=col_map_path, + spec_path=spec_path, + debug=debug, + delimiter='!!', + has_percent=has_percent) + else: + logging.fatal('At the moment, we support only zip files.') + + +if __name__ == '__main__': + flags.DEFINE_string( + 'option', 'all', + 'Specify how to run the process, colmap -- generates column map, ' + 'process -- runs processing, all -- runs colmap first and then ' + 'processing') + flags.DEFINE_string( + 'table_prefix', None, + '[for processing] Subject Table ID as a prefix for output files, ' + 'eg: S2303') + flags.DEFINE_string('spec_path', None, 'Path to the JSON spec [mandatory]') + flags.DEFINE_string( + 'input_path', None, + 'Path to input directory with (current support only for zip files)') + flags.DEFINE_string('output_dir', './', 'Path to the output directory') + flags.DEFINE_boolean( + 'has_percent', False, + '[for processing] Specify if the dataset has percentage values that ' + 'need to be converted to counts') + flags.DEFINE_boolean( + 'debug', False, + '[for processing] Set the flag to add additional columns to debug') + flags.mark_flags_as_required(['table_prefix', 'spec_path', 'input_path']) + app.run(main) diff --git a/scripts/us_census/acs5yr/subject_tables/S2303/process_test.py b/scripts/us_census/acs5yr/subject_tables/S2303/process_test.py index 2faf5ba82f..9e5af17ba3 100644 --- a/scripts/us_census/acs5yr/subject_tables/S2303/process_test.py +++ b/scripts/us_census/acs5yr/subject_tables/S2303/process_test.py @@ -14,9 +14,13 @@ #pylint: disable=import-error _CODEDIR = os.path.dirname(os.path.realpath(__file__)) +sys.path.insert(1, _CODEDIR) sys.path.insert(1, os.path.join(_CODEDIR, '..', 'common')) from generate_col_map import process_zip_file -from .process import S2303SubjectTableDataLoader +try: + from .process import S2303SubjectTableDataLoader, process_subject_tables, set_column_map +except ImportError: + from process import S2303SubjectTableDataLoader, process_subject_tables, set_column_map #pylint: enable=wrong-import-position #pylint: enable=import-error @@ -28,33 +32,25 @@ _EXPECTED_CSV = os.path.join(_TEST_DIR, 'expected.csv') -def set_column_map(input_path, spec_path, output_dir): - generated_col_map = process_zip_file(input_path, - spec_path, - write_output=False) - f = open(os.path.join(output_dir, 'column_map.json'), 'w') - json.dump(generated_col_map, f, indent=4) - f.close() - - class ProcessTest(unittest.TestCase): def test_create_csv_mcf_tmcf(self): with tempfile.TemporaryDirectory() as tmp_dir: - set_column_map(_TEST_DATA_ZIP, _CONFIG_PATH, tmp_dir) - column_map_path = os.path.join(tmp_dir, 'column_map.json') - data_loader = S2303SubjectTableDataLoader( - table_id='S2303', - col_delimiter='!!', - has_percent=True, - debug=False, - output_path_dir=tmp_dir, - json_spec=_CONFIG_PATH, - column_map_path=column_map_path, - decimal_places=3, - estimate_period=5, - header_row=1) - data_loader._process_zip_file(_TEST_DATA_ZIP) + column_map_path = set_column_map(_TEST_DATA_ZIP, + _CONFIG_PATH, + tmp_dir, + table_prefix='S2303') + process_subject_tables(table_prefix='S2303', + input_path=_TEST_DATA_ZIP, + output_dir=tmp_dir, + column_map_path=column_map_path, + spec_path=_CONFIG_PATH, + debug=False, + delimiter='!!', + has_percent=True, + decimal_places=3, + estimate_period=5, + header_row=1) mcf_file_path = os.path.join(tmp_dir, 'S2303_output.mcf') tmcf_file_path = os.path.join(tmp_dir, 'S2303_output.tmcf')