Merge branch 'datacommonsorg:master' into norwayAuto

Author: susmitakumari-art

import-automation/executor/app/executor/import_executor.py

@@ -75,6 +75,8 @@
 
 AUTO_IMPORT_JOB_STAGE = "auto-import-job-stage"
 AUTO_IMPORT_JOB_STATUS = "auto-import-job-status"
+IMPORT_SUMMARY_FILE = "import_summary.json"
+STAGING_PATH = "staging"
 
 
 class ImportStatus(Enum):
@@ -129,6 +131,7 @@ class ImportStatusSummary:
     latest_version: str = ''
     execution_time: int = 0  # seconds
     data_volume: int = 0  # bytes
+    job_id: str = ''
     next_refresh: datetime = None
     """Holds a set of monitoring stats for an import."""
     import_stats: dict = dataclasses.field(default_factory=dict)
@@ -794,11 +797,9 @@ def _update_import_status_table(
             ]
             values = [
                 import_summary.import_name, import_summary.status.name,
-                os.getenv('BATCH_JOB_NAME',
-                          os.getenv('BATCH_JOB_UID', 'local-run')),
-                import_summary.execution_time, import_summary.data_volume,
-                spanner.COMMIT_TIMESTAMP, import_summary.next_refresh,
-                import_summary.latest_version
+                import_summary.job_id, import_summary.execution_time,
+                import_summary.data_volume, spanner.COMMIT_TIMESTAMP,
+                import_summary.next_refresh, import_summary.latest_version
             ]
             # Update import timestamp only if import completed successfully.
             if import_summary.status == ImportStatus.READY:
@@ -813,15 +814,22 @@ def _update_import_status_table(
 
     def _update_latest_version(self, version, output_dir, import_spec,
                                import_summary):
+        if self.config.skip_gcs_upload:
+            logging.warning(
+                "Skipping latest version update as per import config.")
+            return
         logging.info(f'Updating import latest version {version}')
         self.uploader.upload_string(
             version,
             os.path.join(output_dir, self.config.storage_version_filename))
         self.uploader.upload_string(
             self._import_metadata_mcf_helper(import_spec),
             os.path.join(output_dir, self.config.import_metadata_mcf_filename))
+        self.uploader.upload_string(
+            json.dumps(dataclasses.asdict(import_summary), default=str),
+            os.path.join(output_dir, IMPORT_SUMMARY_FILE))
+        # Add current version to the history of versions if import was successful.
         if self.config.storage_version_history_filename:
-            # Add current version to the history of versions.
             history_filename = os.path.join(
                 output_dir, self.config.storage_version_history_filename)
             versions_history = [version]
@@ -847,7 +855,11 @@ def _import_one_helper(
     """
         start_time = time.time()
         import_name = import_spec['import_name']
-        import_summary = ImportStatusSummary(import_name=import_name)
+        job_id = os.getenv('BATCH_JOB_NAME',
+                           os.getenv('BATCH_JOB_UID', 'local-run'))
+        import_summary = ImportStatusSummary(import_name=import_name,
+                                             job_id=job_id)
+
         self.counters.add_counter(f'import-{import_name}', 1)
         urls = import_spec.get('data_download_url')
         if urls:
@@ -935,19 +947,22 @@ def _import_one_helper(
                 import_summary.import_stats.get('mcf_data_size', 0) +
                 import_summary.import_stats.get('validation_data_size', 0))
             logging.info(import_summary)
+            self.uploader.upload_string(
+                json.dumps(dataclasses.asdict(import_summary), default=str),
+                os.path.join(output_dir, version, IMPORT_SUMMARY_FILE))
+
             if self.config.ignore_validation_status or validation_status:
                 import_summary.status = ImportStatus.READY
-                if not self.config.skip_gcs_upload:
-                    self._update_latest_version(version, output_dir,
-                                                import_spec, import_summary)
-                else:
-                    logging.warning(
-                        "Skipping latest version update as per import config.")
             else:
                 logging.error(
-                    "Skipping latest version update due to validation failure.")
+                    "Staging latest version update due to validation failure.")
                 import_summary.status = ImportStatus.VALIDATION
-                self._update_import_status_table(import_summary)
+
+            # Update version and metadata files in staging folder for failed imports
+            version_dir = output_dir if import_summary.status == ImportStatus.READY else os.path.join(
+                output_dir, STAGING_PATH)
+            self._update_latest_version(version, version_dir, import_spec,
+                                        import_summary)
 
         if self.importer:
             self.importer.delete_previous_output(relative_import_dir,

scripts/fbi/hate_crime/aggregations/manifest.json

@@ -20,7 +20,12 @@
             "source_files": [
                 "hate_crime_data/*"
             ],
-            "cron_schedule": "0 10 21 * *"
+            "cron_schedule": "0 10 21 * *",
+            "resource_limits": {
+                "cpu": 16,
+                "memory": 256,
+                "disk": 300
+            }
         }
     ]
-}
+}

scripts/us_bts/latch/manifest.json

@@ -34,7 +34,12 @@
                 "input_files/usbts_tract_household_transportation_2017_input.csv",
                 "input_files/usbts_tract_household_transportation_2009_input.csv"
             ],
-            "cron_schedule": "0 1 6 * *"
+            "cron_schedule": "0 1 6 * *",
+            "resource_limits": {
+                "cpu": 16,
+                "memory": 128,
+                "disk": 300
+            }
         }
     ]
-}
+}

scripts/us_census/pep/population_estimate_by_race/preprocess.py

@@ -517,7 +517,12 @@ def _clean_county_2022_csv_file(df: pd.DataFrame,
             '2': '2020',
             '3': '2021',
             '4': '2022',
-            '5': '2023'
+            '5': '2023',
+            '6': '2024',
+            '7': '2025',
+            '8': '2026',
+            '9': '2027',
+            '10': '2028'
         }
         df = df.replace({'YEAR': conversion_of_year_to_value})
         df.insert(6, 'geo_ID', 'geoId/', True)
@@ -1006,12 +1011,15 @@ def _transform_data(self, df: pd.DataFrame, file_path: str) -> None:
                 "Count_Person_AsianOrPacificIslander",\
                 "Count_Person_TwoOrMoreRaces","Count_Person_NonWhite"]]
             df_before_2000 = self.df[self.df["Year"] < 2000]
-            df_county_after_2000 = self.df[(self.df["Year"] >= 2000) &
-                                           (self.df["geo_ID"] != "country/USA")
-                                           & (self.df["geo_ID"].str.len() > 9)]
+            df_county_after_2000 = self.df[
+                (self.df["Year"] >= 2000) &
+                (self.df["geo_ID"] != "country/USA") &
+                (self.df["geo_ID"].str.len() > 9)].drop_duplicates(
+                    subset=['Year', 'geo_ID'], keep='last')
             df_national_state_2000 = self.df[(self.df["Year"] >= 2000) & (
                 (self.df["geo_ID"].str.len() <= 9) |
-                (self.df["geo_ID"] == "country/USA"))]
+                (self.df["geo_ID"] == "country/USA"))].drop_duplicates(
+                    subset=['Year', 'geo_ID'], keep='last')
             df_before_2000.to_csv(os.path.join(
                 self.cleaned_csv_file_path,
                 "USA_Population_Count_by_Race_before_2000.csv"),

scripts/us_census/pep/us_pep_sex/process.py

@@ -1092,7 +1092,11 @@ def process(self):
                 value_vars=['Count_Person_Male', 'Count_Person_Female'],
                 var_name="SV",
                 value_name="Observation")
-
+            subset_cols = ['Year', 'geo_ID', 'Measurement_Method', 'SV']
+            # 2. Drop duplicates based on those columns, keeping the first occurrence
+            final_df.drop_duplicates(subset=subset_cols,
+                                     keep='first',
+                                     inplace=True)
             final_df.to_csv(self._cleaned_csv_file_path, index=False)
             sv_list = ['Count_Person_Female', 'Count_Person_Male']
             self._generate_mcf(sv_list)
@@ -1117,7 +1121,7 @@ def fetch_skip_urls_from_gcs(GCS_BUCKET_NAME: str,
 
 def is_valid_url(url):
     try:
-        response = requests.get(url, timeout=10)
+        response = requests.get(url, timeout=20)
         if response.status_code != 200:
             return False
         content_type = response.headers.get("Content-Type", "")

scripts/us_nces/demographics/private_school/manifest.json

@@ -24,7 +24,12 @@
                     "cleaned_csv": "gcs_folder/output_place/us_nces_demographics_private_place.csv"
                 }
             ],
-            "cron_schedule": "30 3 1 3,6,9,12 *"
+            "cron_schedule": "30 3 1 3,6,9,12 *",
+            "resource_limits": {
+                "cpu": 32,
+                "memory": 512,
+                "disk": 300
+            }
         }
     ]
-}
+} 

scripts/us_nces/demographics/private_school/process.py

@@ -67,7 +67,7 @@ def set_generate_statvars_flag(self, flag: bool):
 
 if __name__ == '__main__':
     try:
-        logging.set_verbosity(1)
+        logging.set_verbosity(logging.INFO)
         logging.info("Main Method Starts For Private School District ")
         gcs_output_dir_local = os.path.join(
             os.path.dirname(os.path.abspath(__file__)), "gcs_folder")

scripts/us_nces/demographics/public_school/manifest.json

@@ -26,9 +26,9 @@
             ],
             "cron_schedule": "30 3 2 3,6,9,12 *",
             "resource_limits": {
-                "cpu": 8,
-                "memory": 128,
-                "disk": 200
+                "cpu": 32,
+                "memory": 512,
+                "disk": 300
             }
         }
     ]

scripts/us_nces/demographics/public_school/process.py

@@ -53,7 +53,7 @@ class NCESPublicSchool(USEducation):
 
 if __name__ == '__main__':
     try:
-        logging.set_verbosity(1)
+        logging.set_verbosity(logging.INFO)
         logging.info("Main Method Starts For Public School")
         gcs_output_dir_local = os.path.join(
             os.path.dirname(os.path.abspath(__file__)), "gcs_folder")
@@ -98,4 +98,6 @@ class NCESPublicSchool(USEducation):
         logging.info("Main Method Completed For Public School District ")
 
     except Exception as e:
-        logging.fatal(f"Error While Running Public School Process: {e} ")
+        # Only log the first 1000 characters of the error message
+        error_msg = str(e)[:1000]
+        logging.fatal(f"Error While Running Public School Process: {error_msg}")

scripts/us_nces/demographics/school_district/manifest.json

@@ -26,9 +26,9 @@
             ],
             "cron_schedule": "30 3 3 3,6,9,12 *",
             "resource_limits": {
-                "cpu": 8,
-                "memory": 128,
-                "disk": 200
+                "cpu": 32,
+                "memory": 512,
+                "disk": 300
             }
         }
     ]