Status

Motivation

Apache Airflow is primarily designed for time-based and dependency-based scheduling of workflows. However, modern data architectures often require near real-time processing and the ability to react to events from various sources, such as message queues. This proposal aims to introduce native event-driven capabilities to Airflow, allowing users to create workflows that can be triggered by external events, thus enabling more responsive data pipelines.

Proposal

Note. In this AIP we refer “asset” as a dataset in Airflow 2.10. We use “asset” because “dataset” is renamed “asset” in AIP 73.

Today in Airflow, you can build event-based workflows using either external task sensor, sensors/deferrable operators, REST API, and dataset (to be renamed as Assets). Scheduling using assets has recently gained popularity given the efficient execution, monitoring support and API capabilities. An Airflow asset is a logical grouping of data. Upstream producer tasks can update assets, and asset updates contribute to scheduling downstream consumer DAGs.

Example:

from airflow.datasets import Asset
with DAG(...):
    MyOperator(
        # this task updates example.csv
        outlets=[Asset("s3://dataset-bucket/example.csv")],
        ...,
    )
with DAG(
    # this DAG should be run when example.csv is updated
    schedule=[Asset("s3://dataset-bucket/example.csv")],
    ...,
):
    ...

In this example, the first DAG sends an event (or updates) the asset, and the second DAG is scheduled upon asset update.

However, as illustrated in the example above, updating the asset is the user's responsibility. In the example above the user uses a DAG but other techniques such as using the “Create dataset event” Rest API are available to the user to update an asset. All these techniques are great but require some work from the user to set-up a pipeline in order to update the asset.

Ideally, there should be an end to end solution in Airflow to trigger DAGs based on external event such as:

• A file has been created in a storage service
• A database has been updated
• Message comes in a queue

Design

The goal is to build a solution in Airflow to automatically update Assets based on external events. This scheduling can be categorized into two categories:

• Poll based event-driven scheduling
• Push based event-driven scheduling

Poll based event-driven scheduling

Airflow constantly monitors the state of an external resource and updates the asset whenever the external resource reaches a given state (if it does reach it). To achieve this, the plan is to leverage Airflow Triggers. Triggers are small, asynchronous pieces of Python code whose job is to poll an external resource state. Today, triggers are used exclusively for deferrable operators but the goal here would be to use them as well to update assets based on external conditions.

DAG author experience

Below is an example of DAG triggered when a specific file is created in an S3 bucket.

trigger = S3KeyTrigger(
    bucket_name="<my_bucket>",
    bucket_key="<my_file>",
)
asset = Asset("s3://<my_bucket>/<my_file>", watchers=[trigger])

with DAG(
    dag_id=DAG_ID,
    schedule=asset,
    start_date=datetime(2021, 1, 1),
    tags=["example"],
    catchup=False,
):
    empty_task = EmptyOperator(task_id="empty_task")

    chain(empty_task)

Sequence

Below is a simplified version of a sequence diagram describing what is going on in Airflow when DAGs such as above are present in an Airflow environment.

Avoid infinite scheduling

Current triggers implementation is perfectly suited for sensors and deferrable operators but is not compatible with DAG scheduling. The reason is most of the triggers are waiting for an external resource to reach a given state. Examples:

• Wait for a file to exist in a storage service
• Wait for a job to be in a success state
• Wait for a row to be present in a database

Scheduling upon these conditions would lead to infinite scheduling because once the condition is reached, it is very likely it will remain for quite some time. Example: if a DAG is scheduled when a specific job state reaches a success state, when it does, the job state will remain in this state. Therefore, scheduling a DAG using that condition will lead to infinite scheduling when the job reaches this state. Another example, S3KeyTrigger checks if a given file is present in a S3 bucket. Once this specific file is created in the S3 bucket, S3KeyTrigger will always exit successfully since the condition “is the file X present in the bucket Y” is True. In this case, the consequence would be to keep triggering any DAG scheduled based on that trigger every-time the triggerer execute the triggers (every second).

To avoid this infinite scheduling loop, we want to only fire events if we have not done so since the last event update was received. For example, if a S3 file is updated, and we haven’t fired an event since this file updated time, then we trigger the event. As a result, we want to reuse the concept of triggers but we do not want to use the current implementation of triggers specific to deferrable operators and sensors. Therefore, there are two options:

• One trigger implementation should be specific to either deferrable operators/sensors or scheduling
• Introduce a new method in BaseTrigger (e.g. schedule) similar to the existing method run . schedule would be used to check scheduling decisions and run would be used to check defer decisions

Some other parameters could also be added to add control for the DAG author to configure how often a DAG can be scheduled based on these events. See more in the section “Additional considerations (future work)”.

Polling rate

By default the triggerer checks every second all triggers which subsequently call an external API to check the resource state. While it might make sense for deferrable operator and sensors, for poll based scheduling it might be a lot. There a two options:

• Add an optional parameter to BaseTrigger to override this waiting period. Therefore, the DAG author would make the decision on how often a specific trigger poll the external resource
• Set a polling rate specific to triggers used for scheduling. This polling rate could be hardcoded or set by config

Push based event-driven scheduling

As opposed to the poll based event scheduling, the push based event scheduling consists of an event sent from an external system to Airflow whenever this external system detects a change/activity. Examples:

• A user signed up to an external system
• A remote job has been successfully executed
• A file has been created in a storage service

These events are fired by the external service (AWS, Google, ...) to notify such event. The external service needs to be configured to send such event/notification to the Airflow environment. On Airflow side, it needs to receive such event and schedule DAGs that are scheduled upon these events.

To achieve this, here are the main changes we need to introduce in Airflow:

• Create a new HTTP endpoint in Airflow to receive external events. The reason why we chose a HTTP-based notification is most of external services such as AWS, Google can be configured to send notifications to third party application through HTTP. This new endpoint is responsible of receiving all external events from all external services. When configuring the external service to send notification to a third party application, this is the endpoint to send the notification to.
• Create a new base class called BaseEventReceiver (name not definitive, feedbacks are welcome). The event receivers (classes that inherit from BaseEventReceiver) are responsible to parse a notification sent by the external service (a HTTP request) and checks whether this notification matches what is expected. As an example, S3FileCreationEventReceiver would be responsible to parse and check whether a given HTTP request matches the notification sent by AWS when a S3 file is created in a S3 bucket. On AWS side, this notification would be sent by the service EventBridge. These event receivers are used by the new HTTP endpoint to identify the kind of event received by Airflow.

DAG author experience

Below is an example of DAG triggered when a user register to an external system.

event = UserSignUpEventReceiver(...)
dataset = Asset("user_pool", events=[event])

with DAG(
    dag_id=DAG_ID,
    schedule=dataset,
    start_date=datetime(2021, 1, 1),
    tags=["example"],
    catchup=False,
):
    empty_task = EmptyOperator(task_id="empty_task")

    chain(empty_task)

Sequence

Below is a simplified version of sequence diagram describing what is going on in Airflow when DAGs such as above is present in an Airflow environment.

Authentication

The event receivers are also responsible for the authentication with the external service. Notifications sent by the external service will include authentication information that needs to be checked by the event receivers. This is very important because it verifies that the HTTP request is sent by the actual external service.

Where the event receiver endpoint should be?

The event receiver endpoint is merely just a HTTP endpoint with no authentication check (the authentication check happens in event receivers). The plan is to add this endpoint as part of the webserver.

In the codebase, there are multiple options:

1. Expose it as a new new endpoint in the Rest API
2. Create a new endpoint as a new view
3. Create a new API with only one endpoint

I have not yet decided which option I like better. Feedback is appreciated :)

Poll based event scheduling VS push based event scheduling

Why having two different mechanism when one seems “better” than the other. It is true that push based event scheduling is more performant and less costly than the poll based event one. The reason why we want to have both mechanisms is, we might not be able to have a push based event scheduling for all events. Some external resources we want to monitor might not have the option to send a HTTP request whenever a change is detected. In this case, the poll based event scheduling is the best option we have. Also, ease of use and time to configure is shorter with poll based scheduling. Therefore, some users might want to use poll based scheduling for simpleness and others might prefer the more performant push based approach.

Additional considerations (future work)

Airflow is not designed to handle 100s of event per second, and provide below optimization would allow users to tune the behaviour while keeping Airflow scheduling performant

1. Configurable trigger behaviour: Allow DAG authors to specify whether they want the DAG to be triggered:
1. For every event
2. At a specified interval, processing batches of events
3. When a certain number of events have accumulated
2. Batch processing: Support batching of events

Considerations

What problem does it solve?

1. Lack of native support for event-driven workflows, especially push-based event scheduling, in Airflow
2. Difficulty in integrating Airflow with real-time data sources and message queues
3. Inability to trigger DAGs based on external events efficiently

Are there any downsides to this change?

No.

Which users are affected by the change?

Only users who want to use this new feature. This change does not break any existing feature.

How are users affected by the change? (e.g. DB upgrade required?)

DB upgrade is required since some modifications to the DB is needed. These changes are only creation of new DB tables:

• New table to record association between assets and triggers
• New table to record event receivers
• New table to record association between event receivers and assets

What is the level of migration effort (manual and automated) needed for the users to adapt to the breaking changes? (especially in context of Airflow 3)

The migration effort is relatively low for existing workflows, as this proposal introduces new features without breaking existing functionality. However:

1. DAG authors who want to leverage event-driven capabilities will need to modify their DAGs to use the new classes and related concepts
2. Some existing DAGs that use custom solutions for event-driven workflows may need to be refactored to use the new native capabilities

What defines this AIP as "done"?

Poll based scheduling and push based scheduling as described in this AIP handled in Airflow.