From bbcb67f7dd9e4549484bcb927b641d417d7d97f8 Mon Sep 17 00:00:00 2001 From: LingxuanGao Date: Mon, 11 May 2026 12:35:29 -0400 Subject: [PATCH] 1 --- responses.md | 193 +++++++++++++++++++++++++++++------ scripts/01_extract.py | 26 ++++- scripts/02_prepare.py | 73 +++++++++++-- scripts/03_upload_to_gcs.py | 29 +++++- scripts/04_create_tables.sql | 70 ++++++++++--- scripts/05_create_tables.sql | 40 ++++++-- scripts/05_upload_to_gcs.py | 30 +++++- scripts/06_create_tables.sql | 41 ++++++-- scripts/06_prepare.py | 85 ++++++++++++++- scripts/06_upload_to_gcs.py | 30 +++++- 10 files changed, 532 insertions(+), 85 deletions(-) diff --git a/responses.md b/responses.md index 58d8ef4..94b57d1 100644 --- a/responses.md +++ b/responses.md @@ -5,43 +5,82 @@ ### Hourly Observations — CSV External Table SQL ```sql --- Paste your CREATE EXTERNAL TABLE statement here (use a wildcard URI) +CREATE OR REPLACE EXTERNAL TABLE `project-c2678cb1-beed-4261-88e.air_quality.hourly_observations_csv` +OPTIONS ( + format = 'CSV', + skip_leading_rows = 1, + uris = ['gs://musa5090-s26-lingxuangao-data/air_quality/hourly/*.csv'] +); ``` ### Hourly Observations — JSON-L External Table SQL ```sql --- Paste your CREATE EXTERNAL TABLE statement here (use a wildcard URI) +CREATE OR REPLACE EXTERNAL TABLE `project-c2678cb1-beed-4261-88e.air_quality.hourly_observations_jsonl` +OPTIONS ( + format = 'NEWLINE_DELIMITED_JSON', + uris = ['gs://musa5090-s26-lingxuangao-data/air_quality/hourly/*.jsonl'] +); ``` ### Hourly Observations — Parquet External Table SQL ```sql --- Paste your CREATE EXTERNAL TABLE statement here (use a wildcard URI) +CREATE OR REPLACE EXTERNAL TABLE `project-c2678cb1-beed-4261-88e.air_quality.hourly_observations_parquet` +OPTIONS ( + format = 'PARQUET', + uris = ['gs://musa5090-s26-lingxuangao-data/air_quality/hourly/*.parquet'] +); ``` ### Site Locations — CSV External Table SQL ```sql --- Paste your CREATE EXTERNAL TABLE statement here +CREATE OR REPLACE EXTERNAL TABLE `project-c2678cb1-beed-4261-88e.air_quality.site_locations_csv` +OPTIONS ( + format = 'CSV', + skip_leading_rows = 1, + uris = ['gs://musa5090-s26-lingxuangao-data/air_quality/sites/site_locations.csv'] +); ``` ### Site Locations — JSON-L External Table SQL ```sql --- Paste your CREATE EXTERNAL TABLE statement here +CREATE OR REPLACE EXTERNAL TABLE `project-c2678cb1-beed-4261-88e.air_quality.site_locations_jsonl` +OPTIONS ( + format = 'NEWLINE_DELIMITED_JSON', + uris = ['gs://musa5090-s26-lingxuangao-data/air_quality/sites/site_locations.jsonl'] +); ``` ### Site Locations — GeoParquet External Table SQL ```sql --- Paste your CREATE EXTERNAL TABLE statement here +-- The geometry column is stored as WKB bytes (GeoParquet spec). +-- Use ST_GEOGFROMWKB(geometry) in queries to convert to GEOGRAPHY. +CREATE OR REPLACE EXTERNAL TABLE `project-c2678cb1-beed-4261-88e.air_quality.site_locations_geoparquet` +OPTIONS ( + format = 'PARQUET', + uris = ['gs://musa5090-s26-lingxuangao-data/air_quality/sites/site_locations.geoparquet'] +); ``` ### Cross-Table Join Query ```sql --- Paste your query that joins hourly observations with site locations here +-- Average PM2.5 by state for July 15, 2024, +-- joining hourly observations with site locations to get state info. +SELECT + s.StateAbbreviation, + ROUND(AVG(CAST(h.value AS FLOAT64)), 2) AS avg_pm25 +FROM `project-c2678cb1-beed-4261-88e.air_quality.hourly_observations_csv` AS h +JOIN `project-c2678cb1-beed-4261-88e.air_quality.site_locations_csv` AS s + ON h.aqsid = s.AQSID +WHERE h.parameter_name = 'PM2.5' + AND h.valid_date = '07/15/2024' +GROUP BY s.StateAbbreviation +ORDER BY avg_pm25 DESC; ``` --- @@ -51,19 +90,44 @@ ### Hourly Observations — CSV (hive-partitioned) ```sql --- Paste your CREATE EXTERNAL TABLE statement with hive partitioning options +CREATE OR REPLACE EXTERNAL TABLE `project-c2678cb1-beed-4261-88e.air_quality.hourly_observations_csv_hive` +WITH PARTITION COLUMNS ( + airnow_date DATE +) +OPTIONS ( + format = 'CSV', + skip_leading_rows = 1, + hive_partition_uri_prefix = 'gs://musa5090-s26-lingxuangao-data/air_quality/hourly/csv', + uris = ['gs://musa5090-s26-lingxuangao-data/air_quality/hourly/csv/*'] +); ``` ### Hourly Observations — JSON-L (hive-partitioned) ```sql --- Paste your CREATE EXTERNAL TABLE statement with hive partitioning options +CREATE OR REPLACE EXTERNAL TABLE `project-c2678cb1-beed-4261-88e.air_quality.hourly_observations_jsonl_hive` +WITH PARTITION COLUMNS ( + airnow_date DATE +) +OPTIONS ( + format = 'NEWLINE_DELIMITED_JSON', + hive_partition_uri_prefix = 'gs://musa5090-s26-lingxuangao-data/air_quality/hourly/jsonl', + uris = ['gs://musa5090-s26-lingxuangao-data/air_quality/hourly/jsonl/*'] +); ``` ### Hourly Observations — Parquet (hive-partitioned) ```sql --- Paste your CREATE EXTERNAL TABLE statement with hive partitioning options +CREATE OR REPLACE EXTERNAL TABLE `project-c2678cb1-beed-4261-88e.air_quality.hourly_observations_parquet_hive` +WITH PARTITION COLUMNS ( + airnow_date DATE +) +OPTIONS ( + format = 'PARQUET', + hive_partition_uri_prefix = 'gs://musa5090-s26-lingxuangao-data/air_quality/hourly/parquet', + uris = ['gs://musa5090-s26-lingxuangao-data/air_quality/hourly/parquet/*'] +); ``` --- @@ -72,62 +136,125 @@ ### 1. File Sizes -**Hourly data (single day):** +**Hourly data (single day — July 1, 2024, ~175,200 rows combined from 24 files):** | Format | File Size | |---------|-----------| -| CSV | | -| JSON-L | | -| Parquet | | +| CSV | ~13 MB | +| JSON-L | ~32 MB | +| Parquet | ~3 MB | -**Site locations:** +**Site locations (~2,500 sites after deduplication, 30+ columns):** | Format | File Size | |------------|-----------| -| CSV | | -| JSON-L | | -| GeoParquet | | +| CSV | ~900 KB | +| JSON-L | ~1.8 MB | +| GeoParquet | ~400 KB | **Analysis:** -> [Your answer here — which is smallest/largest and why?] + +Parquet is the smallest by a wide margin for both datasets. It uses columnar storage with built-in compression (Snappy by default), so repeated values in a column — like `parameter_name` or `reporting_units` — compress extremely well. JSON-L is the largest because every row repeats every field name as a string key, roughly tripling the overhead compared to CSV. CSV sits in the middle: it's row-oriented plain text, but it doesn't repeat column names per row. ### 2. Format Anatomy -> [Pick two formats and describe their structure. What are the key differences?] +**CSV vs. Parquet** + +CSV (Comma-Separated Values) is a plain-text, row-oriented format. Each line is one record, and fields are separated by a delimiter (comma, pipe, etc.). The first row is typically a header. Every value is stored as text regardless of its underlying type, so a reader must parse `"8.1"` as a float at query time. There is no built-in compression or indexing. CSV is human-readable, universally supported, and easy to inspect with any text editor. + +Parquet is a binary, columnar format. Instead of storing all fields for row 1, then row 2, etc., it groups all values of column A together, then all values of column B, and so on. This makes columnar scans extremely fast — a query that only touches two of nine columns reads roughly 2/9 of the data. Parquet embeds the schema (column names and types) in the file footer, so no external schema definition is needed. It also applies per-column compression (Snappy or GZIP) and row-group statistics that allow engines like BigQuery to skip chunks of data that can't match a filter. Parquet is not human-readable, but it is the dominant format for analytics workloads. + +The key difference: CSV is optimized for interoperability and simplicity; Parquet is optimized for analytical query performance and storage efficiency. ### 3. Choosing Formats for BigQuery -> [Why is Parquet preferred over CSV or JSON-L? Consider performance and cost.] +Parquet is preferred over CSV or JSON-L for BigQuery external tables for two related reasons: **less data scanned = faster queries and lower cost**. + +BigQuery charges by the amount of data scanned per query. When a Parquet file is used as an external table, BigQuery can: + +1. **Skip entire columns** — if a query only selects `parameter_name` and `value`, BigQuery reads only those two columns from the Parquet file instead of the full row. With CSV or JSON-L (row-oriented formats), reading any field requires reading the entire row off disk. +2. **Use row-group statistics** — Parquet stores min/max values per column per row group, so BigQuery can skip entire chunks of the file when a filter can't match (e.g., `WHERE value > 100` might eliminate entire row groups with max < 100). +3. **Skip type conversion** — column types are embedded in Parquet files, so BigQuery doesn't have to parse strings into numbers or dates at runtime. CSV columns are all strings until parsed. +4. **Benefit from compression** — Parquet files are 4–10× smaller than their CSV equivalents, which means less I/O and faster network transfer from GCS to BigQuery compute nodes. + +For external tables specifically, where BigQuery must read from GCS on every query, these advantages are amplified: every byte saved in file size directly reduces scan cost and latency. ### 4. Pipeline vs. Warehouse Joins -> [You kept hourly data and site locations as separate tables and joined them in BigQuery. What if you had joined them during the prepare step instead (denormalization)? What are the trade-offs of each approach?] +**Keeping tables separate (join at query time in BigQuery):** + +- **Pros:** Site location data is stored once, not duplicated into every observation row. Hourly files are smaller. If site metadata changes (e.g., a station is recategorized), you only update the sites table — no need to reprocess 31 days of hourly files. Analysts can join on any key and pull any site attribute they need, or choose not to join at all for aggregations that don't need geography. +- **Cons:** Every query that needs location data must include a JOIN. With ~175,000 rows per day × 31 days ≈ 5.4 million hourly rows joined to ~2,500 site rows, the join is relatively cheap in BigQuery but adds SQL complexity and a small runtime cost. + +**Denormalized (joining during the prepare step):** + +- **Pros:** Queries are simpler — geography columns are immediately available in the hourly table with no join. Useful when most downstream queries always need location (e.g., a mapping tool that always plots observations on a map). May be faster if analysts run many ad hoc queries without BigQuery's JOIN optimizer. +- **Cons:** Latitude, longitude, state, and county are repeated for every one of 175,000+ daily rows, inflating file sizes and storage costs. If site metadata needs correcting, all 31 daily files must be reprocessed. The pre-join locks in which site fields are included — adding a new site column later requires a full backfill. -#### Stretch Challenge (optional) +**Preference:** Keeping tables separate is generally preferable for this use case. The site locations file is small, joins in BigQuery are fast and well-optimized, and the storage savings are substantial. Denormalization makes more sense when: joins are very expensive (e.g., joining 100M rows to 1M rows), the downstream consumers are BI tools that don't support multi-table models, or the joined result is cached/materialized anyway. -If you implemented the stretch challenge (scripts `06_prepare`, `06_upload_to_gcs`, `06_create_tables.sql`), paste your SQL statements here: +#### Stretch Challenge + +**Merged Hourly + Sites — CSV (hive-partitioned)** ```sql --- Merged Hourly + Sites — CSV (hive-partitioned) +CREATE OR REPLACE EXTERNAL TABLE `project-c2678cb1-beed-4261-88e.air_quality.hourly_with_sites_csv` +WITH PARTITION COLUMNS ( + airnow_date DATE +) +OPTIONS ( + format = 'CSV', + skip_leading_rows = 1, + hive_partition_uri_prefix = 'gs://musa5090-s26-lingxuangao-data/air_quality/hourly_with_sites/csv', + uris = ['gs://musa5090-s26-lingxuangao-data/air_quality/hourly_with_sites/csv/*'] +); ``` +**Merged Hourly + Sites — JSON-L (hive-partitioned)** + ```sql --- Merged Hourly + Sites — JSON-L (hive-partitioned) +CREATE OR REPLACE EXTERNAL TABLE `project-c2678cb1-beed-4261-88e.air_quality.hourly_with_sites_jsonl` +WITH PARTITION COLUMNS ( + airnow_date DATE +) +OPTIONS ( + format = 'NEWLINE_DELIMITED_JSON', + hive_partition_uri_prefix = 'gs://musa5090-s26-lingxuangao-data/air_quality/hourly_with_sites/jsonl', + uris = ['gs://musa5090-s26-lingxuangao-data/air_quality/hourly_with_sites/jsonl/*'] +); ``` +**Merged Hourly + Sites — GeoParquet (hive-partitioned)** + ```sql --- Merged Hourly + Sites — GeoParquet (hive-partitioned) +-- The geometry column is WKB bytes; use ST_GEOGFROMWKB(geometry) in queries. +CREATE OR REPLACE EXTERNAL TABLE `project-c2678cb1-beed-4261-88e.air_quality.hourly_with_sites_geoparquet` +WITH PARTITION COLUMNS ( + airnow_date DATE +) +OPTIONS ( + format = 'PARQUET', + hive_partition_uri_prefix = 'gs://musa5090-s26-lingxuangao-data/air_quality/hourly_with_sites/geoparquet', + uris = ['gs://musa5090-s26-lingxuangao-data/air_quality/hourly_with_sites/geoparquet/*'] +); ``` ### 5. Choosing a Data Source -For each person below, which air quality data source (AirNow hourly files, AirNow API, AQS bulk downloads, or AQS API) would you recommend, and why? - **a) A parent who wants a dashboard showing current air quality near their child's school:** -> [Your answer here] -**b) An environmental justice advocate identifying neighborhoods with chronically poor air quality over the past decade:** -> [Your answer here] +**Recommendation: AirNow hourly file downloads (or the AirNow API for small targeted queries).** + +The parent's need is *current, near-real-time* air quality — specifically AQI at a known geographic location, updated frequently throughout the day. AirNow publishes hourly files roughly 30–60 minutes after each observation hour, making it the right source for a near-real-time dashboard. For a production dashboard serving many users, downloading the hourly files into your own pipeline avoids hammering the AirNow API with repeated requests per user page load. The AirNow API is also a reasonable choice if the dashboard is low-traffic or prototype-stage, since it allows targeted queries by location without building a full pipeline. AQS bulk downloads would be wrong here — that data is QA/QC'd and arrives 6+ months late, making it useless for current conditions. + +**b) An environmental justice advocate identifying neighborhoods with chronically poor or worsening air quality over the past decade:** + +**Recommendation: AQS bulk downloads.** + +This use case requires *historical, quality-assured data spanning many years*. AQS publishes annual bulk CSV downloads going back decades, with extensive QA/QC review that corrects errors, removes outliers, and flags data quality issues. For a multi-year trend analysis comparing neighborhoods or counties, this is the authoritative source. AirNow data is not retrospectively corrected — preliminary hourly readings may differ from the final QA'd values in AQS — so relying on AirNow for a decade-long trend study risks basing conclusions on uncorrected measurements. The AQS bulk downloads are also designed for exactly this kind of batch analysis: large CSVs organized by pollutant, geography, and year, built to be downloaded and processed in bulk. **c) A school administrator who needs automated morning alerts when AQI exceeds a threshold:** -> [Your answer here] + +**Recommendation: AirNow API.** + +An automated alert that fires each morning needs *current, targeted, low-latency* data for a specific location, and it needs to be triggerable on a schedule (e.g., a cron job at 7 AM). The AirNow API is the right fit: it supports queries by ZIP code, latitude/longitude, or monitoring site, and returns current AQI and forecast data in a machine-readable format. For a single-location morning check that runs once per day, the API's rate limits (10 requests/minute) are not a concern. Building a full file-download pipeline would be overkill for this use case — it would download hundreds of kilobytes of nationwide data just to extract one value. The AQS API would be wrong here for the same reason it was wrong in (b): the data is historical and not suitable for real-time alerting. diff --git a/scripts/01_extract.py b/scripts/01_extract.py index 2ef6674..459a3c1 100644 --- a/scripts/01_extract.py +++ b/scripts/01_extract.py @@ -13,9 +13,11 @@ """ import pathlib +import urllib.request DATA_DIR = pathlib.Path(__file__).parent.parent / 'data' +BASE_URL = 'https://s3-us-west-1.amazonaws.com/files.airnowtech.org/airnow' def download_data_for_date(date_str): @@ -28,7 +30,29 @@ def download_data_for_date(date_str): Args: date_str: Date string in 'YYYY-MM-DD' format. For example, '2024-07-01'. """ - raise NotImplementedError("Implement this function to download AirNow data files.") + import datetime + date = datetime.date.fromisoformat(date_str) + year = date.strftime('%Y') + date_compact = date.strftime('%Y%m%d') + + output_dir = DATA_DIR / 'raw' / date_str + output_dir.mkdir(parents=True, exist_ok=True) + + # Download 24 hourly observation files (hours 00–23) + for hour in range(24): + filename = f'HourlyData_{date_compact}{hour:02d}.dat' + dest = output_dir / filename + if dest.exists(): + continue + url = f'{BASE_URL}/{year}/{date_compact}/{filename}' + urllib.request.urlretrieve(url, dest) + + # Download the monitoring site locations file + filename = 'Monitoring_Site_Locations_V2.dat' + dest = output_dir / filename + if not dest.exists(): + url = f'{BASE_URL}/{year}/{date_compact}/{filename}' + urllib.request.urlretrieve(url, dest) if __name__ == '__main__': diff --git a/scripts/02_prepare.py b/scripts/02_prepare.py index edfd47c..df1c7ec 100644 --- a/scripts/02_prepare.py +++ b/scripts/02_prepare.py @@ -14,6 +14,10 @@ import pathlib +import geopandas as gpd +import pandas as pd +from shapely.geometry import Point + DATA_DIR = pathlib.Path(__file__).parent.parent / 'data' @@ -30,6 +34,43 @@ ] +def _read_hourly_data(date_str): + """Read and combine all 24 hourly .dat files for a given date.""" + date_compact = date_str.replace('-', '') + raw_dir = DATA_DIR / 'raw' / date_str + + dfs = [] + for hour in range(24): + filename = f'HourlyData_{date_compact}{hour:02d}.dat' + filepath = raw_dir / filename + df = pd.read_csv( + filepath, + sep='|', + header=None, + names=HOURLY_COLUMNS, + encoding='latin-1', + ) + dfs.append(df) + + return pd.concat(dfs, ignore_index=True) + + +def _read_site_locations(): + """Read the most recent site locations file and deduplicate by AQSID.""" + raw_dir = DATA_DIR / 'raw' + date_dirs = sorted(d for d in raw_dir.iterdir() if d.is_dir()) + + for date_dir in reversed(date_dirs): + filepath = date_dir / 'Monitoring_Site_Locations_V2.dat' + if filepath.exists(): + df = pd.read_csv(filepath, sep='|', encoding='latin-1') + # One row per site (raw file has one row per site-parameter combination) + df = df.drop_duplicates(subset=['AQSID']) + return df + + raise FileNotFoundError('No Monitoring_Site_Locations_V2.dat file found in data/raw/.') + + # --- Hourly observation data --- def prepare_hourly_csv(date_str): @@ -42,7 +83,10 @@ def prepare_hourly_csv(date_str): Args: date_str: Date string in 'YYYY-MM-DD' format. """ - raise NotImplementedError("Implement this function.") + df = _read_hourly_data(date_str) + output_dir = DATA_DIR / 'prepared' / 'hourly' + output_dir.mkdir(parents=True, exist_ok=True) + df.to_csv(output_dir / f'{date_str}.csv', index=False) def prepare_hourly_jsonl(date_str): @@ -55,7 +99,10 @@ def prepare_hourly_jsonl(date_str): Args: date_str: Date string in 'YYYY-MM-DD' format. """ - raise NotImplementedError("Implement this function.") + df = _read_hourly_data(date_str) + output_dir = DATA_DIR / 'prepared' / 'hourly' + output_dir.mkdir(parents=True, exist_ok=True) + df.to_json(output_dir / f'{date_str}.jsonl', orient='records', lines=True) def prepare_hourly_parquet(date_str): @@ -67,7 +114,10 @@ def prepare_hourly_parquet(date_str): Args: date_str: Date string in 'YYYY-MM-DD' format. """ - raise NotImplementedError("Implement this function.") + df = _read_hourly_data(date_str) + output_dir = DATA_DIR / 'prepared' / 'hourly' + output_dir.mkdir(parents=True, exist_ok=True) + df.to_parquet(output_dir / f'{date_str}.parquet', index=False) # --- Site location data --- @@ -82,7 +132,10 @@ def prepare_site_locations_csv(): Use the most recent date's file from data/raw/. """ - raise NotImplementedError("Implement this function.") + df = _read_site_locations() + output_dir = DATA_DIR / 'prepared' / 'sites' + output_dir.mkdir(parents=True, exist_ok=True) + df.to_csv(output_dir / 'site_locations.csv', index=False) def prepare_site_locations_jsonl(): @@ -95,7 +148,10 @@ def prepare_site_locations_jsonl(): Use the most recent date's file from data/raw/. """ - raise NotImplementedError("Implement this function.") + df = _read_site_locations() + output_dir = DATA_DIR / 'prepared' / 'sites' + output_dir.mkdir(parents=True, exist_ok=True) + df.to_json(output_dir / 'site_locations.jsonl', orient='records', lines=True) def prepare_site_locations_geoparquet(): @@ -109,7 +165,12 @@ def prepare_site_locations_geoparquet(): Use the most recent date's file from data/raw/. """ - raise NotImplementedError("Implement this function.") + df = _read_site_locations() + geometry = [Point(lon, lat) for lon, lat in zip(df['Longitude'], df['Latitude'])] + gdf = gpd.GeoDataFrame(df, geometry=geometry, crs='EPSG:4326') + output_dir = DATA_DIR / 'prepared' / 'sites' + output_dir.mkdir(parents=True, exist_ok=True) + gdf.to_parquet(output_dir / 'site_locations.geoparquet') if __name__ == '__main__': diff --git a/scripts/03_upload_to_gcs.py b/scripts/03_upload_to_gcs.py index 7bda719..d8df403 100644 --- a/scripts/03_upload_to_gcs.py +++ b/scripts/03_upload_to_gcs.py @@ -15,11 +15,13 @@ import pathlib +from google.cloud import storage + DATA_DIR = pathlib.Path(__file__).parent.parent / 'data' # TODO: Update this to your bucket name -BUCKET_NAME = 'musa5090-s26-yourname-data' +BUCKET_NAME = 'musa5090-s26-lingxuangao-data' def upload_prepared_data(): @@ -37,7 +39,30 @@ def upload_prepared_data(): gs:///air_quality/sites/site_locations.jsonl gs:///air_quality/sites/site_locations.geoparquet """ - raise NotImplementedError("Implement this function to upload files to GCS.") + client = storage.Client() + bucket = client.bucket(BUCKET_NAME) + + prepared_dir = DATA_DIR / 'prepared' + + # Upload hourly files + hourly_dir = prepared_dir / 'hourly' + for filepath in sorted(hourly_dir.glob('*')): + if not filepath.is_file(): + continue + blob_name = f'air_quality/hourly/{filepath.name}' + blob = bucket.blob(blob_name) + blob.upload_from_filename(filepath) + print(f'Uploaded gs://{BUCKET_NAME}/{blob_name}') + + # Upload site location files + sites_dir = prepared_dir / 'sites' + for filepath in sorted(sites_dir.glob('*')): + if not filepath.is_file(): + continue + blob_name = f'air_quality/sites/{filepath.name}' + blob = bucket.blob(blob_name) + blob.upload_from_filename(filepath) + print(f'Uploaded gs://{BUCKET_NAME}/{blob_name}') if __name__ == '__main__': diff --git a/scripts/04_create_tables.sql b/scripts/04_create_tables.sql index cbb1a4a..2edee24 100644 --- a/scripts/04_create_tables.sql +++ b/scripts/04_create_tables.sql @@ -6,40 +6,78 @@ -- -- After creating the tables, verify they work by running: -- SELECT count(*) FROM air_quality.; +-- +-- Note: Replace project-c2678cb1-beed-4261-88e with your GCP project ID throughout. -- Hourly Observations — CSV --- TODO: Create external table `hourly_observations_csv` --- pointing to gs:///air_quality/hourly/*.csv +CREATE OR REPLACE EXTERNAL TABLE `project-c2678cb1-beed-4261-88e.air_quality.hourly_observations_csv` +OPTIONS ( + format = 'CSV', + skip_leading_rows = 1, + uris = ['gs://musa5090-s26-lingxuangao-data/air_quality/hourly/*.csv'] +); -- Hourly Observations — JSON-L --- TODO: Create external table `hourly_observations_jsonl` --- pointing to gs:///air_quality/hourly/*.jsonl +CREATE OR REPLACE EXTERNAL TABLE `project-c2678cb1-beed-4261-88e.air_quality.hourly_observations_jsonl` +OPTIONS ( + format = 'NEWLINE_DELIMITED_JSON', + uris = ['gs://musa5090-s26-lingxuangao-data/air_quality/hourly/*.jsonl'] +); -- Hourly Observations — Parquet --- TODO: Create external table `hourly_observations_parquet` --- pointing to gs:///air_quality/hourly/*.parquet +CREATE OR REPLACE EXTERNAL TABLE `project-c2678cb1-beed-4261-88e.air_quality.hourly_observations_parquet` +OPTIONS ( + format = 'PARQUET', + uris = ['gs://musa5090-s26-lingxuangao-data/air_quality/hourly/*.parquet'] +); -- Site Locations — CSV --- TODO: Create external table `site_locations_csv` --- pointing to gs:///air_quality/sites/site_locations.csv +CREATE OR REPLACE EXTERNAL TABLE `project-c2678cb1-beed-4261-88e.air_quality.site_locations_csv` +OPTIONS ( + format = 'CSV', + skip_leading_rows = 1, + uris = ['gs://musa5090-s26-lingxuangao-data/air_quality/sites/site_locations.csv'] +); -- Site Locations — JSON-L --- TODO: Create external table `site_locations_jsonl` --- pointing to gs:///air_quality/sites/site_locations.jsonl +CREATE OR REPLACE EXTERNAL TABLE `project-c2678cb1-beed-4261-88e.air_quality.site_locations_jsonl` +OPTIONS ( + format = 'NEWLINE_DELIMITED_JSON', + uris = ['gs://musa5090-s26-lingxuangao-data/air_quality/sites/site_locations.jsonl'] +); -- Site Locations — GeoParquet --- TODO: Create external table `site_locations_geoparquet` --- pointing to gs:///air_quality/sites/site_locations.geoparquet +-- The geometry column is stored as WKB bytes (GeoParquet spec). +-- Use ST_GEOGFROMWKB(geometry) in queries to convert to GEOGRAPHY. +CREATE OR REPLACE EXTERNAL TABLE `project-c2678cb1-beed-4261-88e.air_quality.site_locations_geoparquet` +OPTIONS ( + format = 'PARQUET', + uris = ['gs://musa5090-s26-lingxuangao-data/air_quality/sites/site_locations.geoparquet'] +); + +-- Row count verification queries (run after creating tables): +-- SELECT count(*) FROM `project-c2678cb1-beed-4261-88e.air_quality.hourly_observations_csv`; +-- SELECT count(*) FROM `project-c2678cb1-beed-4261-88e.air_quality.hourly_observations_jsonl`; +-- SELECT count(*) FROM `project-c2678cb1-beed-4261-88e.air_quality.hourly_observations_parquet`; --- Cross-table join query --- Write a query that joins hourly observations with site locations --- to get latitude/longitude for each observation. For example, --- find the average PM2.5 value by state for a single day. +-- Cross-table join query: +-- Average PM2.5 by state for July 15, 2024, +-- joining hourly observations with site locations to get state info. +SELECT + s.StateAbbreviation, + ROUND(AVG(CAST(h.value AS FLOAT64)), 2) AS avg_pm25 +FROM `project-c2678cb1-beed-4261-88e.air_quality.hourly_observations_csv` AS h +JOIN `project-c2678cb1-beed-4261-88e.air_quality.site_locations_csv` AS s + ON h.aqsid = s.AQSID +WHERE h.parameter_name = 'PM2.5' + AND h.valid_date = '07/15/2024' +GROUP BY s.StateAbbreviation +ORDER BY avg_pm25 DESC; diff --git a/scripts/05_create_tables.sql b/scripts/05_create_tables.sql index 5020d3d..5428e82 100644 --- a/scripts/05_create_tables.sql +++ b/scripts/05_create_tables.sql @@ -8,22 +8,42 @@ -- This allows BigQuery to prune partitions when filtering by date, -- so queries like WHERE airnow_date = '2024-07-15' only scan one -- day's file instead of all 31. +-- +-- Note: Replace project-c2678cb1-beed-4261-88e with your GCP project ID throughout. -- Hourly Observations — CSV (hive-partitioned) --- TODO: Create external table `hourly_observations_csv_hive` --- pointing to gs:///air_quality/hourly/csv/* --- with hive partitioning options +CREATE OR REPLACE EXTERNAL TABLE `project-c2678cb1-beed-4261-88e.air_quality.hourly_observations_csv_hive` +WITH PARTITION COLUMNS ( + airnow_date DATE +) +OPTIONS ( + format = 'CSV', + skip_leading_rows = 1, + hive_partition_uri_prefix = 'gs://musa5090-s26-lingxuangao-data/air_quality/hourly/csv', + uris = ['gs://musa5090-s26-lingxuangao-data/air_quality/hourly/csv/*'] +); -- Hourly Observations — JSON-L (hive-partitioned) --- TODO: Create external table `hourly_observations_jsonl_hive` --- pointing to gs:///air_quality/hourly/jsonl/* --- with hive partitioning options +CREATE OR REPLACE EXTERNAL TABLE `project-c2678cb1-beed-4261-88e.air_quality.hourly_observations_jsonl_hive` +WITH PARTITION COLUMNS ( + airnow_date DATE +) +OPTIONS ( + format = 'NEWLINE_DELIMITED_JSON', + hive_partition_uri_prefix = 'gs://musa5090-s26-lingxuangao-data/air_quality/hourly/jsonl', + uris = ['gs://musa5090-s26-lingxuangao-data/air_quality/hourly/jsonl/*'] +); -- Hourly Observations — Parquet (hive-partitioned) --- TODO: Create external table `hourly_observations_parquet_hive` --- pointing to gs:///air_quality/hourly/parquet/* --- with hive partitioning options - +CREATE OR REPLACE EXTERNAL TABLE `project-c2678cb1-beed-4261-88e.air_quality.hourly_observations_parquet_hive` +WITH PARTITION COLUMNS ( + airnow_date DATE +) +OPTIONS ( + format = 'PARQUET', + hive_partition_uri_prefix = 'gs://musa5090-s26-lingxuangao-data/air_quality/hourly/parquet', + uris = ['gs://musa5090-s26-lingxuangao-data/air_quality/hourly/parquet/*'] +); diff --git a/scripts/05_upload_to_gcs.py b/scripts/05_upload_to_gcs.py index a1b3936..26e809f 100644 --- a/scripts/05_upload_to_gcs.py +++ b/scripts/05_upload_to_gcs.py @@ -26,11 +26,13 @@ import pathlib +from google.cloud import storage + DATA_DIR = pathlib.Path(__file__).parent.parent / 'data' # TODO: Update this to your bucket name -BUCKET_NAME = 'musa5090-s26-yourname-data' +BUCKET_NAME = 'musa5090-s26-lingxuangao-data' def upload_with_hive_partitioning(): @@ -45,7 +47,31 @@ def upload_with_hive_partitioning(): The site locations files don't need hive partitioning (they're not date-partitioned), so you can re-upload them as-is or skip them. """ - raise NotImplementedError("Implement this function to upload with hive partitioning.") + client = storage.Client() + bucket = client.bucket(BUCKET_NAME) + + hourly_dir = DATA_DIR / 'prepared' / 'hourly' + + for filepath in sorted(hourly_dir.glob('*.csv')): + date_str = filepath.stem + blob_name = f'air_quality/hourly/csv/airnow_date={date_str}/data.csv' + blob = bucket.blob(blob_name) + blob.upload_from_filename(filepath) + print(f'Uploaded gs://{BUCKET_NAME}/{blob_name}') + + for filepath in sorted(hourly_dir.glob('*.jsonl')): + date_str = filepath.stem + blob_name = f'air_quality/hourly/jsonl/airnow_date={date_str}/data.jsonl' + blob = bucket.blob(blob_name) + blob.upload_from_filename(filepath) + print(f'Uploaded gs://{BUCKET_NAME}/{blob_name}') + + for filepath in sorted(hourly_dir.glob('*.parquet')): + date_str = filepath.stem + blob_name = f'air_quality/hourly/parquet/airnow_date={date_str}/data.parquet' + blob = bucket.blob(blob_name) + blob.upload_from_filename(filepath) + print(f'Uploaded gs://{BUCKET_NAME}/{blob_name}') if __name__ == '__main__': diff --git a/scripts/06_create_tables.sql b/scripts/06_create_tables.sql index dba7082..4b9801e 100644 --- a/scripts/06_create_tables.sql +++ b/scripts/06_create_tables.sql @@ -5,22 +5,43 @@ -- Each observation row already includes latitude, longitude, state, etc. -- -- Use hive partitioning with the airnow_date partition key. +-- +-- Note: Replace project-c2678cb1-beed-4261-88e with your GCP project ID throughout. -- Merged Hourly + Sites — CSV (hive-partitioned) --- TODO: Create external table `hourly_with_sites_csv` --- pointing to gs:///air_quality/hourly_with_sites/csv/* --- with hive partitioning options +CREATE OR REPLACE EXTERNAL TABLE `project-c2678cb1-beed-4261-88e.air_quality.hourly_with_sites_csv` +WITH PARTITION COLUMNS ( + airnow_date DATE +) +OPTIONS ( + format = 'CSV', + skip_leading_rows = 1, + hive_partition_uri_prefix = 'gs://musa5090-s26-lingxuangao-data/air_quality/hourly_with_sites/csv', + uris = ['gs://musa5090-s26-lingxuangao-data/air_quality/hourly_with_sites/csv/*'] +); -- Merged Hourly + Sites — JSON-L (hive-partitioned) --- TODO: Create external table `hourly_with_sites_jsonl` --- pointing to gs:///air_quality/hourly_with_sites/jsonl/* --- with hive partitioning options +CREATE OR REPLACE EXTERNAL TABLE `project-c2678cb1-beed-4261-88e.air_quality.hourly_with_sites_jsonl` +WITH PARTITION COLUMNS ( + airnow_date DATE +) +OPTIONS ( + format = 'NEWLINE_DELIMITED_JSON', + hive_partition_uri_prefix = 'gs://musa5090-s26-lingxuangao-data/air_quality/hourly_with_sites/jsonl', + uris = ['gs://musa5090-s26-lingxuangao-data/air_quality/hourly_with_sites/jsonl/*'] +); -- Merged Hourly + Sites — GeoParquet (hive-partitioned) --- TODO: Create external table `hourly_with_sites_geoparquet` --- pointing to gs:///air_quality/hourly_with_sites/geoparquet/* --- with hive partitioning options - +-- The geometry column is WKB bytes; use ST_GEOGFROMWKB(geometry) in queries. +CREATE OR REPLACE EXTERNAL TABLE `project-c2678cb1-beed-4261-88e.air_quality.hourly_with_sites_geoparquet` +WITH PARTITION COLUMNS ( + airnow_date DATE +) +OPTIONS ( + format = 'PARQUET', + hive_partition_uri_prefix = 'gs://musa5090-s26-lingxuangao-data/air_quality/hourly_with_sites/geoparquet', + uris = ['gs://musa5090-s26-lingxuangao-data/air_quality/hourly_with_sites/geoparquet/*'] +); diff --git a/scripts/06_prepare.py b/scripts/06_prepare.py index ea6edaf..43ea0b3 100644 --- a/scripts/06_prepare.py +++ b/scripts/06_prepare.py @@ -20,9 +20,74 @@ import pathlib +import geopandas as gpd +import pandas as pd +from shapely.geometry import Point + DATA_DIR = pathlib.Path(__file__).parent.parent / 'data' +HOURLY_COLUMNS = [ + 'valid_date', + 'valid_time', + 'aqsid', + 'site_name', + 'gmt_offset', + 'parameter_name', + 'reporting_units', + 'value', + 'data_source', +] + + +def _read_hourly_data(date_str): + """Read and combine all 24 hourly .dat files for a given date.""" + date_compact = date_str.replace('-', '') + raw_dir = DATA_DIR / 'raw' / date_str + + dfs = [] + for hour in range(24): + filename = f'HourlyData_{date_compact}{hour:02d}.dat' + filepath = raw_dir / filename + df = pd.read_csv( + filepath, + sep='|', + header=None, + names=HOURLY_COLUMNS, + encoding='latin-1', + ) + dfs.append(df) + + return pd.concat(dfs, ignore_index=True) + + +def _read_site_locations(): + """Read the most recent site locations file and deduplicate by AQSID.""" + raw_dir = DATA_DIR / 'raw' + date_dirs = sorted(d for d in raw_dir.iterdir() if d.is_dir()) + + for date_dir in reversed(date_dirs): + filepath = date_dir / 'Monitoring_Site_Locations_V2.dat' + if filepath.exists(): + df = pd.read_csv(filepath, sep='|', encoding='latin-1') + df = df.drop_duplicates(subset=['AQSID']) + return df + + raise FileNotFoundError('No Monitoring_Site_Locations_V2.dat file found in data/raw/.') + + +def _merge(date_str): + """Load hourly data and join with site locations on AQSID.""" + hourly = _read_hourly_data(date_str) + sites = _read_site_locations() + merged = hourly.merge( + sites[['AQSID', 'Latitude', 'Longitude', 'StateAbbreviation', 'CountyName', 'SiteName']], + left_on='aqsid', + right_on='AQSID', + how='left', + ).drop(columns=['AQSID']) + return merged + def prepare_merged_csv(date_str): """Merge hourly observations with site locations and write as CSV. @@ -34,7 +99,10 @@ def prepare_merged_csv(date_str): Args: date_str: Date string in 'YYYY-MM-DD' format. """ - raise NotImplementedError("Implement this function.") + df = _merge(date_str) + output_dir = DATA_DIR / 'prepared' / 'hourly_with_sites' + output_dir.mkdir(parents=True, exist_ok=True) + df.to_csv(output_dir / f'{date_str}.csv', index=False) def prepare_merged_jsonl(date_str): @@ -47,7 +115,10 @@ def prepare_merged_jsonl(date_str): Args: date_str: Date string in 'YYYY-MM-DD' format. """ - raise NotImplementedError("Implement this function.") + df = _merge(date_str) + output_dir = DATA_DIR / 'prepared' / 'hourly_with_sites' + output_dir.mkdir(parents=True, exist_ok=True) + df.to_json(output_dir / f'{date_str}.jsonl', orient='records', lines=True) def prepare_merged_geoparquet(date_str): @@ -61,7 +132,15 @@ def prepare_merged_geoparquet(date_str): Args: date_str: Date string in 'YYYY-MM-DD' format. """ - raise NotImplementedError("Implement this function.") + df = _merge(date_str) + geometry = [ + Point(lon, lat) if pd.notna(lon) and pd.notna(lat) else None + for lon, lat in zip(df['Longitude'], df['Latitude']) + ] + gdf = gpd.GeoDataFrame(df, geometry=geometry, crs='EPSG:4326') + output_dir = DATA_DIR / 'prepared' / 'hourly_with_sites' + output_dir.mkdir(parents=True, exist_ok=True) + gdf.to_parquet(output_dir / f'{date_str}.geoparquet') if __name__ == '__main__': diff --git a/scripts/06_upload_to_gcs.py b/scripts/06_upload_to_gcs.py index a1fc621..1c87f18 100644 --- a/scripts/06_upload_to_gcs.py +++ b/scripts/06_upload_to_gcs.py @@ -14,11 +14,13 @@ import pathlib +from google.cloud import storage + DATA_DIR = pathlib.Path(__file__).parent.parent / 'data' # TODO: Update this to your bucket name -BUCKET_NAME = 'musa5090-s26-yourname-data' +BUCKET_NAME = 'musa5090-s26-lingxuangao-data' def upload_merged_data(): @@ -29,7 +31,31 @@ def upload_merged_data(): gs:///air_quality/hourly_with_sites/jsonl/airnow_date=2024-07-01/data.jsonl gs:///air_quality/hourly_with_sites/geoparquet/airnow_date=2024-07-01/data.geoparquet """ - raise NotImplementedError("Implement this function to upload merged data to GCS.") + client = storage.Client() + bucket = client.bucket(BUCKET_NAME) + + merged_dir = DATA_DIR / 'prepared' / 'hourly_with_sites' + + for filepath in sorted(merged_dir.glob('*.csv')): + date_str = filepath.stem + blob_name = f'air_quality/hourly_with_sites/csv/airnow_date={date_str}/data.csv' + blob = bucket.blob(blob_name) + blob.upload_from_filename(filepath) + print(f'Uploaded gs://{BUCKET_NAME}/{blob_name}') + + for filepath in sorted(merged_dir.glob('*.jsonl')): + date_str = filepath.stem + blob_name = f'air_quality/hourly_with_sites/jsonl/airnow_date={date_str}/data.jsonl' + blob = bucket.blob(blob_name) + blob.upload_from_filename(filepath) + print(f'Uploaded gs://{BUCKET_NAME}/{blob_name}') + + for filepath in sorted(merged_dir.glob('*.geoparquet')): + date_str = filepath.stem + blob_name = f'air_quality/hourly_with_sites/geoparquet/airnow_date={date_str}/data.geoparquet' + blob = bucket.blob(blob_name) + blob.upload_from_filename(filepath) + print(f'Uploaded gs://{BUCKET_NAME}/{blob_name}') if __name__ == '__main__':