Simplified ETL pipeline with:
source_db -> FastAPI -> httpx ETL -> SQLAlchemy -> target_db
The ETL does not connect directly to source_db. It reads source data through the FastAPI connector:
ETL -> httpx -> FastAPI -> source_db
The project seeds source.data with reproducible random data:
Period: 2024-01-01 00:00:00 to 2024-01-10 23:59:00
Frequency: 1 minute
Rows: 10 × 24 × 60 = 14,400
Source table:
source.data
- timestamp
- wind_speed
- power
- ambient_temprature
Target schema:
target.signal
- id
- name
target.data
- timestamp
- signal_id
- value
The ETL aggregates wind_speed and power into 10-minute intervals using:
mean
min
max
std
Each variable + aggregation is stored as a signal:
wind_speed_mean
wind_speed_min
wind_speed_max
wind_speed_std
power_mean
power_min
power_max
power_std
Expected target rows for one full day:
144 intervals × 2 variables × 4 aggregations = 1,152 rows
cp .env.example .env
docker compose down -v
docker compose up -d source_db target_db
docker compose run --build --rm app python scripts/init.py
docker compose up -d api
docker compose run --rm app python scripts/checks.py all --date 2024-01-02
docker compose run --rm app python scripts/run_etl.py 2024-01-02
docker compose run --rm app python scripts/checks.py targetFastAPI docs:
http://localhost:8000/docs
Dagster orchestrates the ETL as a daily partitioned asset.
For one day, source data is extracted through the API, aggregated, and loaded into target_db.
A partition is a slice of an asset. This ETL is daily, so one partition equals one date.
Run Dagster:
docker compose up -d dagster
docker compose logs -f dagsterCheck Dagster definitions import:
docker compose run --rm app python -c "from dagster_project.definitions import defs; print(defs)"Dagster UI:
http://localhost:3000
Materialize asset:
daily_aggregated_signals
partition: 2024-01-02
After materializing one partition, check target:
docker compose run --rm app python scripts/checks.py target