Airbyte connections and Dagster software-defined assets#
An Airbyte connection defines a series of data streams which are synced between a source and a destination. During a sync, a replica of the data from each data stream is written to the destination, typically as one or more tables. Dagster represents each of the replicas generated in the destination as a software-defined asset. This enables you to easily:
Visualize the streams involved in an Airbyte connection and execute a sync from Dagster
Define downstream computations which depend on replicas produced by Airbyte
Track historical metadata and logs for each data stream
Track data lineage through Airbyte and other tools
The first step in using Airbyte with Dagster is to tell Dagster how to connect to your Airbyte instance using an Airbyte resource. This resource contains information on where the Airbyte instance is located and any credentials needed to access it.
from dagster_airbyte import airbyte_resource
airbyte_instance = airbyte_resource.configured({"host":"localhost","port":8000,# If using basic auth, include username and password:"username":"airbyte","password":{"env":"AIRBYTE_PASSWORD"},})
If you're running Airbyte locally using docker-compose, the host and port parameters should be set to localhost and 8000, respectively. The default basic auth credentials are a username airbyte and password password.
If you're hosting Airbyte externally, you'll need to provide a hostname where the Airbyte webapp and API are accssible, typically on port 80. For more information on the configuration options available for the Airbyte resource, see the API reference.
Step 2: Loading Airbyte asset definitions into Dagster#
The easiest way to get started using Airbyte with Dagster is to have Dagster automatically generate asset defintions from your Airbyte project. This can be done in one of two ways:
Loading Airbyte asset definitions from an Airbyte instance#
To load Airbyte assets into Dagster from a live Airbyte instance, you will need to supply the Airbyte resource that we defined above in step 1. Here, the Airbyte instance is treated as the source of truth.
from dagster_airbyte import load_assets_from_airbyte_instance
# Use the airbyte_instance resource we defined in Step 1
airbyte_assets = load_assets_from_airbyte_instance(airbyte_instance)
The load_assets_from_airbyte_instance function retrieves all of the connections you have defined in the Airbyte interface, creating software-defined assets for each data stream. Each connection has an associated op which triggers a sync of that connection.
Loading Airbyte asset definitions from YAML config#
To load Airbyte assets into Dagster from a set of YAML configuration files, specify the Octavia project directory, which contains the sources, destinations, and connections subfolders. This is the directory where you first ran octavia init. Here, the YAML files are treated as the source of truth for building Dagster assets.
from dagster_airbyte import load_assets_from_airbyte_project
airbyte_assets = load_assets_from_airbyte_project(
project_dir="path/to/airbyte/project",)
The load_assets_from_airbyte_project function parses the YAML metadata, generating a set of software-defined assets which reflect each of the data streams synced by your connections. Each connection has an associated op which triggers a sync of that connection.
Assets loaded from Airbyte require an airbyte_resource, which defines how to connect and interact with your Airbyte instance.
We can add the Airbyte resource we configured above to our Airbyte assets by doing the following:
from dagster_airbyte import load_assets_from_airbyte_project
from dagster import with_resources
# Use the airbyte_instance resource we defined in Step 1
airbyte_assets = with_resources([load_assets_from_airbyte_project(project_dir="path/to/airbyte/project")],{"airbyte": airbyte_instance},)
Instead of having Dagster automatically create the asset defintions for your Airbyte instance, you can opt to individually build them. First, determine the connection IDs for each of the connections you would like to build assets for. The connection ID can be seen in the URL of the connection page when viewing the Airbyte UI.
Then, supply the connection ID and the list of tables which the connection creates in the destination to build_airbyte_assets:
from dagster_airbyte import build_airbyte_assets
airbyte_assets = build_airbyte_assets(
connection_id="87b7fe85-a22c-420e-8d74-b30e7ede77df",
destination_tables=["releases","tags","teams"],)
Manually built Airbyte assets require an airbyte_resource, which defines how to connect and interact with your Airbyte instance.
We can add the Airbyte resource we configured above to our Airbyte assets by doing the following:
from dagster_airbyte import build_airbyte_assets
from dagster import with_resources
airbyte_assets = with_resources(
build_airbyte_assets(
connection_id="87b7fe85-a22c-420e-8d74-b30e7ede77df",
destination_tables=["releases","tags","teams"],),# Use the airbyte_instance resource we defined in Step 1{"airbyte": airbyte_instance},)
Once you have loaded your Airbyte assets into Dagster, you can create assets which depend on them. These can be other assets pulled in from external sources such as dbt or assets defined in Python code.
In this case, we have an Airbyte connection which is storing data in the stargazers table in our Snowflake warehouse. We specify the output IO manager to tell downstream assets how to retreive the data.
Once you have Airbyte assets, you can define a job that runs some or all of these assets on a schedule, triggering the underlying Airbyte sync.
from dagster import ScheduleDefinition, define_asset_job, repository, AssetSelection
# materialize all assets in the repository
run_everything_job = define_asset_job("run_everything", selection="*")# only run my_airbyte_connection and downstream assets
my_etl_job = define_asset_job("my_etl_job", AssetSelection.groups("my_airbyte_connection").downstream())@repositorydefmy_repo():return[
airbyte_assets,
ScheduleDefinition(
job=my_etl_job,
cron_schedule="@daily",),
ScheduleDefinition(
job=run_everything_job,
cron_schedule="@weekly",),]