This project contains a dagster repository that extracts, cleans and loads PUDL's usage metrics into a data warehouse for analysis.
This is the project structure generated by the dagster cli:
| Name | Description |
|---|---|
README.md |
A description and guide for this code repository |
workspace.yaml |
A file that specifies the location of the user code for Dagit and the Dagster CLI |
src/usage_metrics/ |
A Python directory that contains code for your Dagster repository |
tests/ |
A Python directory that contains tests for usage_metrics |
setup.py |
A build script with Python package dependencies for this code repository |
We use pixi to specify and update our development environment.
pixi install
The ETL uses ipinfo to geocode ip addresses.
Grab the ipinfo token by logging in using the credentials saved in our Bitwarden Shared Inframundo Logins collection.
The ETL will look for this token in the IPINFO_TOKEN environment variable.
The DATA_DIR environment variable is required for local development.
In this directory, the script will cache input data and save the processed database.
Dagster stores run logs and caches in a directory stored in the DAGSTER_HOME environment variable.
The usage_metrics/dagster_home/dagster.yaml file contains configuration for the dagster instance.
Note: The usage_metrics/dagster_home/storage directory could grow to become a couple GBs because all op outputs for every run are stored there.
You can read more about the dagster_home directory in the dagster docs.
To use the Kaggle API, sign up for a Kaggle account.
Then go to the 'Account' tab of your user profile and select 'Create API Token'.
This will trigger the download of kaggle.json, a file containing your API credentials.
Use this file to automatically set your Kaggle API credentials locally, or manually set KAGGLE_USER and KAGGLE_KEY environment variables.
To set these environment variables, run these commands:
export IPINFO_TOKEN="{your_token_here}"
export DAGSTER_HOME="$(pwd)/dagster_home/"
export DATA_DIR="$(pwd)/data/" # Required for local development. Input and output data will be saved here.
export KAGGLE_USER="{your_kaggle_username_here}" # If setting manually
export KAGGLE_KEY="{your_kaggle_api_key_here}" # If setting manually
You can also drop these commands into a file and use envrc or similar.
Ask the project admin of the catalyst-cooperative-pudl to add your email to the pudl-usage-metrics-etl group to acquire the adequate permissions to run the ETL locally.
Once you have been added to the group, run:
gcloud auth application-default login
in your terminal. This command will prompt you to log in to your gmail account. Once completed, your Google credentials will be available in your environment.
Git hooks let you automatically run scripts at various points as you manage your source code. "Pre-commit" hook scripts are run when you try to make a new commit. These scripts can review your code and identify bugs, formatting errors, bad coding habits, and other issues before the code gets checked in. This gives you the opportunity to fix those issues before publishing them.
To make sure they are run before you commit any code, you need to enable the pre-commit hooks scripts with this command:
pre-commit install
The scripts that run are configured in the .pre-commit-config.yaml file.
Now the environment is all set up and we can start up dagster!
In your DAGSTER_HOME folder, add a dagster.yaml file or edit your existing one to contain the following code:
run_queue:
max_concurrent_runs: 4
Set this higher or lower based on your available cores and memory. When running backfills, it will prevent you from kicking off 80 concurrent runs.
In one terminal window start the dagster-daemon and UI for local development by running these commands:
pixi run update-local
The dagster-webserver is a long-running service required for schedules, sensors and run queueing. The usage metrics ETL requires the daemon because the data is processed in partitions. Dagster kicks off individual runs for each partition which are sent to a queue managed by the dagster-daemon.
This command simultaneously starts the dagit UI.
This will launch dagit at http://localhost:3000/ if port 3000 is free.
If port 3000 is unavailable, it will pick a random free port, and tell you which one it picked.
If you wish to specify the exact port it will use, you can set it explicitly by running:
pixi run METRICS_PROD_ENV="local" dagster dev -m usage_metrics.etl -p {another_cool_port}
Dagster allows you to kick off
backfills
and run partitions with specific configuration.
There is a job in the usage_metrics/etl sub package for each datasource
(e.g datasette logs, github metrics…).
Each job module contains a series of assets that extract, transform and load the data.
- When run locally, the IO manager will write data to local partitioned parquet files for development.
- When run in production (prod) mode or on Github actions, the IO manager will write data to Google Cloud Storage. If you configure a production run for a previously-existing partition, it will overwrite what is already there. Be sure this is what you want before you kick off the run!
You can run the ETL via the dagit UI or the dagster CLI.
To run a a complete backfill from the Dagit UI go to the job's partitions tab. Then click on the "Launch Backfill" button in the upper left corner of the window. This should bring up a new window with a list of partitions. Click "Select All" and then click the "Submit" button. This will submit a run for each partition. You can follow the runs on the "Runs" tab.
The choice between local development (written to local storage)
and production development (written to cloud storage)
is determined through the METRICS_PROD_ENV environment variable.
By default, you will develop locally using pixi run update-local.
To output to production cloud storage, run the following:
pixi run update-prod
Production runs will write partitions to parquet files in cloud storage.
The load-metrics GitHub action is responsible for updating each parquet with freshly-built partitions.
If you configure a run for a previously-existing partition, it will overwrite what is already there.
Be sure this is what you want before you kick off the run!
If a new column is added or data is processed in a new way, you'll have to run a complete backfill to overwrite all previous partitions.
By default prod runs will write to metrics.catalyst.coop.
You can examine the contents of the default bucket at
https://console.cloud.google.com/storage/browser/metrics.catalyst.coop.
To update a different path or bucket, set GCS_BUCKET:
export GCS_BUCKET="metrics.catalyst.coop/my-cool-folder"
The ETL uses ipinfo for geocoding the user ip addresses which provides 50k free API requests a month.
The usage_metrics.helpers.geocode_ip() function using joblib
to cache API calls so we don't call the API multiple times for a single ip address.
The first time you run the ETL no API calls will be cached so the geocode_ips() op will take a while to complete.
To add a new data source to the dagster repo,
add new modules to the raw and core and out directories and add these modules to the corresponding jobs.
Once the dataset has been tested locally,
run a complete backfill for the job with pixi run update-prod to populate the Cloud Storage bucket.
