[RFC] Airlift: An Airflow Migration Toolkit

[schrockn]

Airlift: An Airflow Migration Toolkit

We’re happy to announce an early preview of Airlift (tutorial), a toolkit to accelerate, lower the cost, and reduce the risk of migrating from Airflow to Dagster.

It facilitates a three-step, incremental process:

• Peer: With a single line of code, observe Airflow DAGs and run history within Dagster
• Observe: Model the lineage of assets orchestrated by Airflow within Dagster and track their execution, without modifying Airflow code.
• Migrate: With minimal changes to Airflow code (until deletion), invoke Dagster via an API from within Airflow. This is configurable on a per-task basis, enabling per-task migration in any order, while keeping Airflow DAG structure and execution history intact during the process.

We’ve seen early success with the approach, and are looking for further feedback and additional design partners. We are specifically looking for high-conviction users who actively want to migrate, but have been stymied by the absence of a readily available incremental approach and tooling support.

Context

Many teams and engineers look to Dagster as their orchestrator of choice. Frequently these teams have an existing orchestrator (or orchestrators). And often that is Airflow.

System evolution and migration is difficult work. The essential quality of a successful migration is incremental process. Incremental process implies a number of critical qualities that make it preferable to “big bang” migrations in nearly all cases:

• Lower risk: Subdividing a complex process into small, undo-able steps dramatically reduces risk.
• Delivers incremental value: The process should be designed to deliver incremental value during the process, which drives momentum and demonstrates value of the investment to stakeholders.
• Lower coordination costs: An incremental process in nearly all cases leads to lower coordination costs as the work of different teams in the system can be decoupled.
• Interruptible. As urgent tasks come up (as they inevitably do) the migration can be “put on ice” with the system in a stable state while those urgent issues are addressed. Incremental process means the entire system is in a valid, operable state at every intermediate point.

Airlift is designed with the above in mind.

Introducing Airlift

We propose a three stage, tool-assisted process for migrating an Airflow instance to Dagster:

• Peer: Airflow DAGs automatically appear in Dagster
• Observe: Observe assets produced by Airflow in the Dagster UI, with zero changes to Airflow code.
• Migrate: Migrate incrementally to Dagster on a task-by-task basis

We’ll go through each of these stages in greater detail and describe the benefits.

Step 1: Peer

The first stage is to “peer” the Airflow instance to Dagster. This requires just a few lines of Dagster code.

from dagster import Definitions
from dagster_airlift.core import (
    AirflowInstance,
    BasicAuthBackend,
    build_defs_from_airflow_instance,
)

airflow_instance = AirflowInstance(
    auth_backend=BasicAuthBackend(
        webserver_url="http://localhost:8080", username="u", password="pwd"
    ),
    name="instance_name",
)

defs = build_defs_from_airflow_instance(airflow_instance=airflow_instance)

Once you’ve added this code to your Dagster deployment, all of your Airflow DAGs will show up automatically as Dagster assets. Any successful runs of Airflow DAGs will register as a materialization.

This effortlessly starts the process of using Dagster as your single control and observability plane. With the DAGs ingested into the asset graph you can, for example:

• Trigger downstream compute based on DAG completion state.
• Add asset checks to these DAGs to validate the results.
• Monitor Airflow DAGs from multiple Airflow instances in a single surface.
• Use alerting in Dagster+.
• Setup freshness checks.

We’ve found that this step is important psychologically for all stakeholders, as they can see, with minimal effort, how Dagster will be used as a universal observability plane during the process.

It also demonstrates the parallelization available throughout this process. Teams interested in building downstream greenfield Dagster assets can do so immediately, completely separate from the “observe” and “migrate” processes happening upstream. This is the power of incrementalism.

Step 2: Observe

The next step of the process is to incrementally add asset lineage to the Dagster instance for the assets orchestrated within Airflow.

Inside traditional Airflow DAGs there exists an implicit, unencoded asset graph that corresponds to physical assets in durable storage. The “observe” step can be thought of as making this implicit graph explicit.

For example in this simple example is an implicit five node asset graph, orchestrated by the three tasks in this DAG:

We model this asset graph in Dagster, and map them to tasks in the Airflow DAG using Airlift’s assets_with_task_mappings function:

from dagster import Definitions
from dagster_dbt import dbt_assets, DbtCliResource
from dagster_airlift.core import (
    build_defs_from_airflow_instance,
    assets_with_task_mappings
)

@dbt_assets(manifest="path/to/manifest.json")
def jaffle_shop_assets(context: AssetExecutionContext, dbt: DbtCliResource):
    yield from dbt.cli(["build"], context=context).stream()
    
  
defs = build_defs_from_airflow_instance(
    airflow_instance=airflow_instance,
    defs=Definitions(
        assets=assets_with_task_mappings(
            # task "build_dbt_models" in "my_dag" orchestrates a dbt project
            dag_id="my_dag",
            task_mappings={"build_dbt_models": [jaffle_shop_assets]},
        ),
        resources={"dbt": DbtCliResource(project_dir="path/to/dbt_project.yml")},
    ),
)

The Dagster UI will now show us the asset relationships within this task, and how they relate to the Airflow DAG. Using the task mapping information, Dagster will mark assets as “materialized” when the corresponding task in Airflow successfully completes.

Orchestrated In Airflow:

Observed in Dagster:

Performing this step before migration allows you to integrate Airflow-orchestrated assets into your global asset graph. You can trace dependencies from other assets Dagster orchestrates, within Airflow instances (so-called “cross DAG dependencies") or even dependencies between multiple Airflow instances. And most functionality will work out of the box for these assets, such as alerting, asset checks, and cataloging features.

Step 3: Migrate

The next stage of Airlift is to actually migrate compute. Up until now, we haven’t touched the Airflow instance at all (beyond ensuring that we can hit its REST API). Now we start that process.

This first step is to change your Airflow instance to enable proxying to Dagster. Proxying has a special meaning in Airlift: swapping out tasks at runtime to invoke the Dagster API. You add a call to proxying_to_dagster at the bottom of any file that the Airflow scheduler imports DAGs from.

from dagster_airlift.in_airflow import (
    proxying_to_dagster, 
    load_proxied_state_from_yaml
)

# In your airflow dags directory
my_dag = DAG(dag_id="my_dag", ...)
# Rest of existing DAG construction

# Add these two lines of code to your dag file, otherwise unaltered.
proxied_state = load_proxied_state_from_yaml("path/to/proxied_state")
proxying_to_dagster(global_vars=globals(), proxied_state=proxied_state)

This function post processes the Airflow DAG and swaps in a task that invokes Dagster programmatically instead of the original task. It does this on a per-task basis based on configuration in your Airflow repo.

On the Dagster side, there must be code that executes the equivalent business logic as the Airflow operator. This typically means there is a factory function that creates Dagster definitions per operator type in your Airflow installation.

For example, imagine you used this pattern for running dbt projects:

args = f"--project-dir path/to/project}"
run_dbt_model = BashOperator(
    task_id="build_dbt_models", 
    bash_command=f"dbt build {args} --project_dir", 
    dag=dag
)

For each one of those, you would use the equivalent dagster code.

# dbt_assets invocation for each project
# If you heavily customized this and had multiple dbt projects you would
# write a function the returns the decorated function
@dbt_assets(
    manifest="path/to/project/target/manifest.json",
    project=DbtProject("path/to/project"),
)
def dbt_project_assets(context: AssetExecutionContext, dbt: DbtCliResource):
    yield from dbt.cli(["build"], context=context).stream()

The effort required to migrate here scales mostly with the number of operators rather than the number of tasks as per-task migration is mechanical and goes quickly. We also provide out-of-the-box operators for many common operators, such as the KubernetesPodOperator, PythonOperator, and BashOperator for DBT.

Once you have a corresponding set of assets for your task, you can “proxy” execution to Dagster, meaning that instead of executing the business logic of your asset in Airflow, it will execute in Dagster via an API call. This is managed by per-dag yaml files in your Airflow repository.

And construct a simple yaml file:

# In a file proxied_state/my_dag.yaml
tasks:
  - id: build_dbt_models
    proxied: True

And now this task will instead materialize a set of Dagster assets. At runtime, the task is replace with a task the invokes Dagster over an API.

This configuration-based approach for per-task proxying confers a number of advantages

• It is mechanical, straightforward, and repeatable.
• It is self-documenting. You can inspect a file for a particular dag and know what is migrated versus not.
• It is undoable. If there is a bug, just switch the flag back to false.
• It can be done in any order. You can start with the “easiest” tasks to build confidence and parallelize the process.
• While it is happening, Airflow DAG structure and execution remains stable. From the standpoint of an observer (whether the be a human or a tool), very little has changed except for the operator type and logs.

Now, let’s say something went wrong with the migration - your downstream check fails or the asset fails to materialize. With a one-line change to your migration file, you can switch execution back to your original Airflow business logic:

# In a file proxied_state/my_dag.yaml
tasks:
  - id: build_dbt_models
    proxied: False

Step 4: Complete the migration

Once an Airflow DAG is fully migrated, you can just add the equivalent schedule to Dagster and then delete the assets_with_task_mappings call and all associated code.

from dagster import Definitions
from dagster_dbt import dbt_assets, DbtCliResource

@dbt_assets(manifest="path/to/manifest.json")
def jaffle_shop_assets(context: AssetExecutionContext, dbt: DbtCliResource):
    yield from dbt.cli(["build"], context=context).stream()
    
  
defs = Definitions(
    assets=[jaffle_shop_assets],
    resources={"dbt": DbtCliResource(project_dir="path/to/dbt_project.yml"),
)

What about existing approaches?

There already exists a package called dagster-airflow which enables Dagster to directly import and execute unmodified Airflow DAGs. This package is best described as an adapter, and is not a complete solution to the problem.

• Because of the norms of Airflow project structure, it involved refactoring Airflow code to put DAG creation inside functions, rather than the normative approach of a single file in global scope.
• Importing unmodified DAGs did not take advantage of Dagster’s asset-oriented features and programming model that enable better development ergonomics and testability.
• Complex DAGs that relied on a globally available Airflow instance DB or advanced Airflow features were not directly executable.

In the end the dagster-airflow was best suited for the scenario where a data platform team had a “long tail” of simple DAGs not under active development. This allowed that team to migrate those DAGs en masse and then de-commission their Airflow infrastructure to reduce maintenance overhead. Something more flexible and incremental like Airlift is required for the general use case.

Future Roadmap

Our initial target for this toolkit are high-conviction users that seek to fully migrate from Airflow to Dagster. However, this is only the first step.

For customers at larger scale or older organizations, there will always be a heterogeneous scheduling landscape for multiple reasons:

• There are legacy orchestration instances that cannot or should not be migrated.
• Scheduling that interacts with the data platform will occur in other mechanisms, such as CI/CD.
• There will always be built-in cron and simple scheduling in adjacent hosted systems.

We view Airlift as the first step towards a generalized orchestration federation solution. You could already see the hints of that in this discussion, as we mention that you are visualize multiple airflow instances in a single Dagster deployment, and that the “observe” step on its own provides incremental value. This can generalize across other orchestrators and contexts.

With this capability, Dagster will land as an asset-oriented control plane that, in addition to providing a best-in-class orchestrator for greenfield data pipelines, also can incorporate and unify existing orchestrators in a single tool.

To that end we improve observability capabilities and we will expand beyond Airflow to incorporate other orchestrators and schedulers as well, such as AWS Step Functions, Azure Data Factory, and Dbt Cloud. And beyond that we plan on delivering the framework and patterns those are built on so that users and the ecosystem can build integrations for other orchestrators and schedulers.

Next Steps

Thanks for reading! In terms of next steps:

• Provide feedback on the approach. Feel free to leave comments here!
• The library and the tutorial are live! Please try it out and give us feedback.
• Become a design partner! We are actively developing this feature and looking to partner with teams directly. The toolkit is still in the preview phase, so we recommend that anyone wishing to deploy this at scale reach out to us. We have created the #airflow-migration in slack so please follow along there.

[albert849]

Nice love this

[sungchun12]

Really really thoughtful

[NicolasPA]

Impressive forward-thinking, as usual from Dagster!

My little nitpick is that I find the namings of the Peer and Observe stages are confusing, it is written as the beginning that Peer means observe too. Maybe it's ok to have longer, but more descriptive titles.
Also, the 3 stages become 4 steps in the details, maybe that's 4 stages, or 2 stages with each 2 steps?

1. Observe Airflow DAGs in Dagster
2. Observe Airflow tasks as Dagster assets
3. Migrate execution of Airflow tasks to Dagster
4. Migrate Airflow DAG scheduling to Dagster

It will be interesting to see Airflift able to migrate from Prefect in the future.

[dpeng817]

Yea I think this is a good point that both peer and observe are technically performing "observation". I think one point in favor of the existing naming convention is that a big part of the "peer" step is also just setting up the connection to your Airflow instance, getting rest API credentialing figured out. Thinking of it as dagster "running alongside" your Airflow instance - hence peer. But I think the naming you suggested here would fit well for our tutorial, even if I bias towards shorter names for memorization purposes.

Regarding Prefect - I think we would be able to do this for any arbitrary python scheduler. We could certainly build that if there's interest!

Thanks 😄