Skip to main content

Dagster Guide

OSO uses Dagster to perform all data orchestration in our backend infrastructure. Most of the time, contributors will be working in Dagster to connect new data sources. If the source data already exists in OSO, take a look at contributing models via sqlmesh.

In order to setup Dagster in your development environment, check out the getting started guide.

Data Architecture​

As much as possible, we store data in an Iceberg cluster, making it the center of gravity of the OSO data lake. Thus, data ingest will typically go into Iceberg tables. pyoso is the best way to query any data in these tables. This will make it significantly easier to decentralize OSO infrastructure in the future.

Job schedules​

All automated job schedules can be found on our public Dagster dashboard.

Currently, our main data pipeline runs once per week on Sundays.

Alert system​

Dagster alert sensors are configured in warehouse/oso_dagster/oso_dagster/factories/alerts.py

Right now, alerts are reported to #alerts in the OSO Discord server.

Secrets Management​

When you are creating new data sources for OSO, you may need to handle secrets (e.g. passwords, access keys, DB connection strings).

warning

DO NOT CHECK SECRETS INTO THE REPOSITORY!

Instead, please use the OSO SecretResolver to properly handle your secrets.

Local secrets​

While you are developing your code, the right place to store secrets is in your root .env file. The OSO SecretResolver organizes secrets by (prefix, group, key). Dagster will automatically load secrets from your environment by the following convention PREFIX__GROUP__KEY. By default, all secrets in Dagster use the prefix, dagster.

For example, we store the OSO app database's connection string under the postgres group. Thus, this is the environment variable we'd set for it:

DAGSTER__POSTGRES__CONNECTION_URL=

You can reference a secret using SecretReference and resolve it using SecretResolver.

from ..utils import SecretReference, SecretResolver

connection_url_ref = SecretReference(group_name="postgres", key="connection_url")
connection_url = secrets.resolve_as_str(connection_url_ref)

In order to get a reference to the SecretResolver, you'll want to accept it as a Dagster resource, and let a resource factory resolve the secret before handing a plain value to the resource itself. The resource stays free of any knowledge of secrets:

# warehouse/oso_dagster/oso_dagster/resources/postgres.py
class PostgresResource(ConfigurableResource):
    """Resource for interacting with Supabase's PostgreSQL."""

    connection_url: str = Field(description="PostgreSQL connection URL.")
# warehouse/oso_dagster/oso_dagster/definitions/resources.py
@resource_factory("oso_app_db")
@time_function(logger)
def oso_app_db_factory(secrets: SecretResolver) -> PostgresResource:
    """Factory function to create a Postgres DB resource."""
    return PostgresResource(
        connection_url=secrets.resolve_as_str(
            SecretReference(group_name="postgres", key="connection_url")
        ),
    )

You can see resources.py and postgres.py as an example, or op_atlas.py for a SecretReference used directly by an asset.

Production secrets​

When you are ready to run your assets in production, please reach out to the core OSO team on Discord. We will arrange a secure way to share your secrets into our production keystore.

Restating SQLMesh models​

To learn how to restate SQLMesh models, check the SQLMesh ops guide.

Seed Data​

To test the integration between dagster assets and SQLMesh models, check the Seed Data section