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 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 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.
You can configure this resource and add it to your Airbyte assets by doing the following:
from dagster_airbyte import airbyte_resource, load_assets_from_airbyte_project
from dagster import with_resources
airbyte_assets = with_resources([load_assets_from_airbyte_project(project_dir="path/to/airbyte/project")],{"airbyte": airbyte_resource.configured({"host":"localhost","port":"8000",# If using basic auth, include username and password:"username":"airbyte","password":{"env":"AIRBYTE_PASSWORD"},})},)
Loading Airbyte asset definitions from an Airbyte instance#
To load Airbyte assets into Dagster from a live Airbyte instance, you will need to configure an Airbyte resource which defines how to connect to that instance. Here, the Airbyte instance is treated as the source of truth.
from dagster_airbyte import airbyte_resource, load_assets_from_airbyte_instance
airbyte_instance = airbyte_resource.configured({"host":"localhost","port":"8000",# If using basic auth, include username and password:"username":"airbyte","password":{"env":"AIRBYTE_PASSWORD"},})
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.
You don't need to use with_resources to bind an Airbyte resource to the assets produced by load_asset_from_airbyte_instance; the Airbyte instance you supply to the function will automatically be bound to them.
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.
You can configure this resource and add it to your Airbyte assets by doing the following:
from dagster_airbyte import airbyte_resource, 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"],),{"airbyte": airbyte_resource.configured({"host":"localhost","port":"8000",# If using basic auth, include username and password:"username":"airbyte","password":{"env":"AIRBYTE_PASSWORD"},})},)