Page History

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 message has been pushed in a message queue such as Google Pubsub, Amazon SQS, Azure Message Bus, Apache Kafka, ...
• A file has been created in a storage service
• A database has been updatedMessage 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

Only the poll based event-driven scheduling is considered as part of this AIP. Some investigation has been done on the push based event-driven scheduling without leading to a satisfying solution. Therefore, it has been decided to focus on the poll based approach while continuing investigating on the push based event-driven scheduling. If it is decided later to move on with a push based event-driven scheduling (this proposal or another), it will be part of another AIP.

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.

Today you could achieve this by using continuous scheduling and having one deferrable operator in the DAG you are continuously scheduling. he purpose of this solution is to achieve the same results without having one DAG that is continuously scheduled to run the deferrable operator.

DAG author experience

Below is an example of DAG triggered when a specific file is created in an S3 bucketnew message is pushed to an Amazon SQS queue.

trigger = S3KeyTrigger SqsSensorTrigger(sqs_queue="<my_queue>")
asset = Asset("<my_queue>", watchers=[trigger])

with DAG(
    bucket_name="<my_bucket>",dag_id=DAG_ID,
    schedule=asset,
    start_date=datetime(2021, 1, 1),
    tags=["example"],
    catchup=False,
):
    bucket_keyempty_task = EmptyOperator(task_id="<myempty_file>task")

    chain(empty_task)

Below is an example of DAG triggered when a given S3 file is updated.

trigger = S3FileUpdatedTrigger(s3_file="<s3_file_uri>"),
)
asset = Asset("s3://<my_bucket>/<my_file><s3_file_uri>", 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)

The only concern of this solution is you do need to be careful with the trigger implementation. The asset will be updated whenever the trigger succeed, so we need to be sure the trigger will succeed only once after the file has been updated. If the trigger keeps succeeding once the file is updated, it will lead to infinite asset updates which leads to infinite DAG executions ...

Note. In the two examples above, I used the the actual resource identifier to monitor as asset name but the asset name can be any string, as such you can use any string to name these assets.

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 secondthis interval is usually configurable in the trigger implementation).

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

We should mention that some triggers are compatible with both use cases. As an example, SqsSensorTrigger  can be used to defer a DAG but also to schedule a DAG since it is waiting for new message in a queue, poll it when there is one and then delete it. For this reason, among the two options listed above, we might want to chose the first one.

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)”.

Out of scope

Items below are out of scope of this AIP. However, I decided to keep them in this document because they might be useful to initiate a conversation if someone decides to work on one of these items once this AIP is completed.

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

Some other features could also be built based upon this AIP. Example of such features: archive and replay of events. However, this is not part of the AIP and considered as a follow-up item.

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
• An OpenLineage event has been emitted

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 create asset event API endpoint of the Airflow environment. On Airflow side, it needs to receive such event and schedule DAGs that are scheduled upon these eventsthis API endpoint needs to handle unknown incoming requests and convert it to valid create asset event API request.

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.

Image Removed

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?

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

• Add an optional mapping mechanism in the create asset event API endpoint between the user (credentials used to call the create asset event API) and a method to convert the request coming from the external service to a valid create asset event API request. Example:
  • Step 1. As a user I configure in my cloud provider to send an event whenever a remote job has been successfully executed. As part of this configuration, I specify the create asset event API endpoint of my Airflow environment as HTTP target. I also configure the event to use specific credentials I created in Airflow for this event (e.g. remote_job_success_user). When this step is done, whenever my remote job is successfully executed, my cloud provider will send an HTTP request to create asset event API endpoint of my Airflow environment.
  • Step 2. As a user I associate a function to the user I used to call my Airflow environment (e.g. remote_job_success_user). This function would be responsible of converting the HTTP request sent by my cloud provider when a remote job is successfully executed to a valid create asset event API request. This function can be written by the user themself or use some of the functions available in providers (these functions do not exist today in Airflow providers). How this association between a user and a function will be done by the user will be determined during the implementation. We can think of using Airflow CLI to help the user make such association but this decision will be delayed to the implementation.

Note. The user might want to handle this conversion themself by using Lambda or any other computing layer between the event and the Airflow environment. This is possible today and will remain possible.

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.

Image Added

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 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.