Skip to main content

Setup

info

Airlift v2 is under active development. You may encounter feature gaps, and the APIs may change. To report issues or give feedback, please reach out to your CSM.

Install dg​

To install dg, follow the dg installation guide.

Create a components-ready project​

To create a components-ready project, follow the project creation guide.

Add the Airlift component type to your environment​

uv add dagster-airlift[core]

Create a new instance of the Airlift component​

Currently dagster-airlift only supports basic authentication against an Airflow instance. You can scaffold a new component into your project using the dg scaffold command:

dg scaffold dagster_airlift.core.components.AirflowInstanceComponent airflow --name my_airflow --auth-type basic_auth

This will create a new component definition in your project under the defs/airflow directory.

tree src/my_project/defs
src/my_project/defs
├── __init__.py
└── airflow
    └── component.yaml

2 directories, 2 files

By default, the component pulls values from environment variables (AIRFLOW_WEBSERVER_URL, AIRFLOW_USERNAME, and AIRFLOW_PASSWORD). While you should never include your password directly in this file, you can update the generated file to hard code the values for the webserver url and username if desired.

cat src/my_project/defs/airflow/component.yaml
type: dagster_airlift.core.components.AirflowInstanceComponent

attributes:
  name: my_airflow
  auth:
    type: basic_auth
    webserver_url: '{{ env("AIRFLOW_WEBSERVER_URL") }}'
    username: '{{ env("AIRFLOW_USERNAME") }}'
    password: '{{ env("AIRFLOW_PASSWORD") }}'

Once this is done, Dagster will do a few things:

  1. All of your airflow dags will be automatically represented in dagster in the "jobs" page. Any jobs pulled from airflow will have an airflow icon to distinguish them
  2. Any airflow datasets will be automatically represented in dagster as assets
  3. Any time an airflow dag executes, that run will be represented

(maybe explain the fact that it creates a sensor called your_airlift_instance__airflow_monitoring_job_sensor that will be responsible for looking for runs in your airflow instance and pulling them in to dagster)