OpenAQ Ingestor

Data ingestion pipeline for OpenAQ that processes air quality measurements from various providers into the OpenAQ database.

Overview

The ingestor handles three main types of data:

1. LCS Pipeline Data (lcs-etl-pipeline/) - Low-cost sensor measurements and metadata
2. Realtime Data (realtime-gzipped/) - Real-time air quality measurements
3. Metadata (lcs-etl-pipeline/stations/) - Sensor station metadata

Architecture

Event-Driven Processing

The ingestor runs as an AWS Lambda function triggered by:

• S3 Events via SNS (production) - Automatically processes new files uploaded to S3
• EventBridge/CloudWatch Events (cron) - Scheduled batch processing
• Direct invocation (manual) - For debugging or reprocessing specific files

Processing Flow

S3 Upload → SNS Topic → Lambda Handler → Load to Staging → ETL to Permanent Tables
                           ↓
                      fetchlogs tracking

Data Pipeline: lcsV2 Process

The lcsV2 module is the primary ingestion pipeline for low-cost sensor data. It processes JSON, NDJSON, and CSV files containing sensor metadata and measurements.

Process Steps

1. File Selection (load_fetchlogs)

• Function: utils.load_fetchlogs(pattern, limit, ascending)
• Purpose: Selects files from fetchlogs table for processing
• Filters:
  • Pattern match (regex)
  • init_datetime IS NOT NULL (initialized records only)
  • completed_datetime IS NULL (unprocessed)
  • has_error = false (no previous errors)
  • Not loaded in last 30 minutes (prevents duplicate processing)
• Updates: Sets loaded_datetime, increments jobs counter, assigns batch_uuid

2. Data Loading (IngestClient.load_key)

• Purpose: Reads file from S3 or local filesystem and parses data
• Supported Formats:
  • JSON - Structured metadata and measurements
  • NDJSON - Newline-delimited JSON (realtime data)
  • CSV - Tabular sensor data
• Populates:
  • self.nodes - Sensor locations/stations
  • self.systems - Sensor systems/instruments
  • self.sensors - Individual sensors
  • self.measurements - Measurement values
  • self.flags - Data quality flags

3. Staging Data Dump (IngestClient.dump)

3a. Locations Dump (dump_locations)

• SQL File: temp_locations_dump.sql
• Purpose: Creates staging tables and writes location data
• Tables Created:
  • staging_keys - Tracks files being processed
  • staging_sensornodes - Node/location data with unique constraints on ingest_id and (source_name, source_id)
  • staging_sensorsystems - System/instrument data linked to nodes
  • staging_sensors - Sensor data linked to systems
  • staging_flags - Data quality flags
• Updates: Sets fetchlogs.loaded_datetime = clock_timestamp() and last_message = 'load_data'

3b. Measurements Dump (dump_measurements)

• SQL File: temp_measurements_dump.sql
• Purpose: Creates staging table and writes measurement data
• Table Created:
  • staging_measurements - Raw measurement values with sensor references
• Data Transfer: Uses PostgreSQL COPY for efficient bulk loading

4. ETL Processing

4a. Process Nodes/Systems/Sensors

• SQL File: etl_process_nodes.sql
• Purpose: Moves data from staging tables to permanent tables
• Operations:
  • Matches nodes to existing sensor_nodes records or creates new ones
  • Links systems to nodes and creates sensor_systems records
  • Links sensors to systems and creates sensors records
  • Handles geometry, timezones, countries lookups
  • Resolves measurands and units
  • Merges metadata JSON fields
• Tables Updated:
  • sensor_nodes
  • sensor_systems
  • sensors

4b. Process Measurements

• SQL File: etl_process_measurements.sql
• Purpose: Processes measurements into permanent storage
• Operations:
  • Links measurements to sensors via ingest_id or (source_name, source_id)
  • Validates and filters measurements
  • Inserts into measurements table
  • Handles duplicates and rejects
  • Creates rollup data for aggregations
• Tables Updated:
  • measurements
  • rejects (invalid measurements)

5. Completion

• Updates: Sets fetchlogs.completed_datetime = clock_timestamp() and last_message = NULL
• Cleanup: Drops staging tables (if using TEMP TABLEs)

SQL Files Reference (lcsV2)

File	Purpose	When Used	Key Operations
temp_locations_dump.sql	Create staging tables for locations	dump_locations() start	CREATE [TEMP] TABLE for nodes, systems, sensors, flags
etl_process_nodes.sql	ETL staging → permanent tables	dump_locations() end	INSERT/UPDATE sensor_nodes, sensor_systems, sensors
temp_measurements_dump.sql	Create staging table for measurements	dump_measurements() start	CREATE [TEMP] TABLE staging_measurements
etl_process_measurements.sql	ETL measurements → permanent table	dump_measurements() end	INSERT measurements, handle rejects, create rollups

Data Pipeline: Realtime Process

The realtime process (fetch.py) handles high-frequency measurement data in NDJSON format.

SQL Files Reference (Realtime)

File	Purpose	When Used
fetch_staging.sql	Create staging table for realtime data	Start of realtime processing
fetch_copy.sql	Copy data into staging	After data loaded
fetch_ingest_full.sql	Process staging → measurements table	Final ETL step

Orchestration

Automated Processing (cronhandler)

Triggered by EventBridge on a schedule (e.g., every 5 minutes):

def cronhandler(event, context):
    # Process metadata (stations)
    load_metadata_db(limit=250, ascending=False)

    # Process realtime measurements
    load_db(limit=50, ascending=False)

    # Process pipeline measurements
    load_measurements_db(limit=250, ascending=False)

Each function:

1. Calls load_fetchlogs() with specific pattern
2. Processes returned records
3. Returns count of processed files

Manual Processing

For debugging or reprocessing specific files:

# Process specific file by ID
load_measurements_file(fetchlogs_id=123)

# Process by key pattern
load_measurements_pattern(pattern='lcs-etl-pipeline/measures/2024%')

# Process batch by UUID
load_measurements_batch(batch_uuid='abc123')

Database Schema

Tracking Table: fetchlogs

Tracks all files processed through the system:

Column	Type	Purpose
fetchlogs_id	int	Primary key (auto-generated)
key	text	S3 object key (unique)
last_modified	timestamptz	S3 object last modified time
init_datetime	timestamptz	When record was initialized (required for automated processing)
loaded_datetime	timestamptz	When file was last loaded for processing
completed_datetime	timestamptz	When processing completed successfully
has_error	boolean	Whether file has processing errors
last_message	text	Status message or error details
jobs	int	Number of times file has been processed
batch_uuid	text	UUID linking files processed together
file_size	bigint	File size in bytes

Staging Tables (Temporary)

Created per-session during processing:

• staging_keys - Files being processed
• staging_sensornodes - Node locations
• staging_sensorsystems - Sensor systems
• staging_sensors - Individual sensors
• staging_flags - Quality flags
• staging_measurements - Raw measurements

Permanent Tables

Final destination for processed data:

• sensor_nodes - Sensor locations/stations
• sensor_systems - Sensor system/instrument metadata
• sensors - Individual sensor configurations
• measurements - Time-series measurement data
• rejects - Invalid/rejected measurements for review

Configuration

Environment Variables

Configured via .env.local (local) or environment (production):

# Database
DATABASE_WRITE_URL=postgresql://user:pass@host:port/database

# S3
FETCH_BUCKET=openaq-fetches

# Processing
USE_TEMP_TABLES=true  # Use temp tables for staging (production=true)
INGEST_TIMEOUT=300    # Timeout for cron processing (seconds)
PAUSE_INGESTING=false # Emergency pause flag

# Limits (records per batch)
PIPELINE_LIMIT=250    # LCS pipeline measurements
REALTIME_LIMIT=50     # Realtime measurements
METADATA_LIMIT=250    # Station metadata

# Order
FETCH_ASCENDING=false # Process newest files first (false) or oldest first (true)

Settings

See ingest/settings.py for full configuration options.

Testing

Run All Tests

poetry run pytest tests/ -v
## increase the log level
poetry run pytest tests/ --log-cli-level=DEBUG

Test Categories

Handler Tests (test_handler_integration.py):

• SNS-wrapped S3 events (production path)
• Direct S3 events (manual invocations)
• EventBridge cron triggers

Cronhandler Tests (test_cronhandler_unit.py):

• Loader orchestration
• Timeout handling
• Error resilience
• Event parameter overrides

Load Fetchlogs Tests (test_load_fetchlogs.py):

• Pattern matching
• Ordering (ascending/descending)
• 30-minute loaded_datetime logic
• Duplicate prevention

lcsV2 Integration Tests (test_lcsv2_integration.py):

• End-to-end data loading
• Staging table verification
• Data integrity checks
• Multiple data formats (JSON, NDJSON, CSV)

Test with Coverage

poetry run pytest tests/ --cov=ingest --cov-report=term-missing

Run Specific Test

poetry run pytest tests/test_handler_integration.py::TestHandlerSNSEvents::test_handler_sns_event_single_file -v

Development Workflow

cli.py — unified ingest tool

A single command-line tool for loading, checking, and downloading fetchlog files. Replaces the older local.py, check.py, and test_ingest.py scripts.

Basic usage

./cli.py [SELECTOR] [ACTION] [OPTIONS]

You must provide one key selector. By default the tool processes each key through the ingest client and rolls back — pass --commit to persist.

Selecting keys

Flag	Purpose
--id ID	single fetchlog id
--batch UUID	all keys in a batch
--pattern REGEX	query fetchlogs by PostgreSQL regex
--s3-prefix PREFIX	list S3 directly under a prefix
--from-file PATH	read keys from a file (one per line, # for comments)
KEY [KEY ...]	positional S3 keys or local paths

Selectors are mutually exclusive. Positional keys are upserted into fetchlogs on the target DB.

Actions

Exactly one of these determines what happens to each key:

Flag	Behavior
(default)	parse, dump to staging, run ETL, report
--preview	parse only; no DB writes at all
--dry-run	resolve keys and print them; no processing
--download [PATH]	download files locally; no DB interaction

Modifiers

Flag	Purpose
--stage-only	dump to staging tables, skip the ETL load step
--keep-staging	use permanent staging tables (overrides USE_TEMP_TABLES) so you can inspect state after a run
--source-db DSN	alternate DB connection string for fetchlog lookups (e.g. query production, load into dev)
--bucket NAME	override FETCH_BUCKET
--limit N	cap on keys processed (default 300)

Reporting

Flag	Purpose
(default)	prints a per-key summary table with matched/added counts
--no-report	suppress the summary table
--csv PATH	write per-key stats to CSV
--diagnose QUERY[,QUERY,...]	run named diagnostic queries; use list to see available names, all to run every one
--diagnose-dir DIR	write diagnostic query results to CSV files in this directory

Diagnostic queries run inside the same transaction as the ingest, so they see uncommitted staging data even without --keep-staging.

Transaction control

Flag	Purpose
(default)	rollback all changes
--commit	persist changes to the target DB

Examples

# Look at a file's parsed contents without touching the DB
./cli.py --key measures/2024-03/foo.json.gz --preview

# See which keys a pattern resolves to, don't process
./cli.py --pattern 'measures/2024-03/.*' --dry-run --limit 50

# Download a production file for local inspection
./cli.py --key measures/2024-03/foo.json.gz --download ~/inspect/

# Load a batch into dev with rollback (default) and full report
./cli.py --batch abc-123

# Same, but persist
./cli.py --batch abc-123 --commit

# Stage-only load with CSV report
./cli.py --from-file keys.txt --stage-only --csv report.csv

# Cross-DB: query prod fetchlogs, load into local dev
./cli.py --pattern 'measures/2024.*' \
    --source-db postgresql://user@prod-host/openaq \
    --commit

# Debug a matching issue with diagnostic queries
./cli.py --key measures/foo.json.gz \
    --diagnose new-nodes-with-nearby,rejects \
    --diagnose-dir ./diag/

# Full inspection: keep staging around, run all diagnostics
./cli.py --key measures/foo.json.gz \
    --keep-staging --diagnose all --diagnose-dir ./diag/

Two-step "look then commit" workflow

Because rollback is the default, a common pattern is:

# First run: inspect what would happen
./cli.py --key foo --diagnose summary

# Second run: same command with --commit
./cli.py --key foo --diagnose summary --commit

Available diagnostic queries

Run ./cli.py --diagnose list to see the full list. Queries are grouped into packs for convenience:

• summary — quick health check (staged counts, sources, units, rejects)
• matching — node matching analysis (unmatched, nearby existing, cross-provider)
• rejects — reject breakdown by cause
• units — sensor/measurement unit reconciliation
• quality — flagged values, suspicious timestamps, stuck sensors
• spatial — geographic proximity and collision detection

Use pack names anywhere a query name is expected:

./cli.py --key foo --diagnose matching,rejects

Processing a Test File with Python API

1. Place test file in the ingest directory or provide path
2. Run with Python:

from ingest.lcsV2 import IngestClient
from datetime import date

client = IngestClient()
client.load_key('path/to/test.json', fetchlogs_id=1, last_modified=str(date.today()))
client.dump(load=True)

Debugging a Failed File

from ingest.utils import get_logs_from_ids

# Get error details
logs = get_logs_from_ids([fetchlogs_id])
print(logs)

# Reprocess with error handling
from ingest.lcsV2 import load_measurements_file
load_measurements_file(fetchlogs_id)

Checking Processing Status

from ingest.utils import load_errors_summary, load_errors_list

# Summary of recent errors
errors = load_errors_summary(days=7)

# List of recent error details
error_list = load_errors_list(limit=10)

Deployment

The ingestor runs as an AWS Lambda function deployed via CDK (see ../cdk/lambda_ingest_stack.py).

Lambda Configuration

• Trigger: SNS topic subscribed to S3 bucket notifications
• Schedule: EventBridge rule for cron processing
• Timeout: Configured based on processing requirements
• Memory: Sufficient for large file processing
• VPC: Access to RDS PostgreSQL database

Deployment

cd ../cdk
cdk deploy lambda-ingest-stack

Monitoring

CloudWatch Metrics

Custom metrics published:

• Files processed per run
• Processing duration
• Error counts
• Record counts

Logs

Check CloudWatch Logs for:

• File processing details
• SQL execution notices
• Error tracebacks
• Performance metrics

Troubleshooting

Common Issues

File not processing:

• Check fetchlogs.init_datetime IS NOT NULL
• Verify has_error = false
• Check if completed_datetime IS NULL
• Ensure not loaded in last 30 minutes

Staging table not found:

• Verify USE_TEMP_TABLES setting
• Check if processing completed (temp tables dropped)
• For debugging, set USE_TEMP_TABLES=false

Duplicate records:

• Check unique constraints on staging tables
• Verify batch_uuid tracking
• Review 30-minute duplicate prevention logic

Memory issues:

• Reduce batch limits (PIPELINE_LIMIT, etc.)
• Increase Lambda memory allocation
• Process files individually with load_measurements_file()

Support

For issues or questions:

• Check test files for examples
• Review SQL files for ETL logic
• See function docstrings for parameter details