Airflow conceptsAdvanced

Event-driven scheduling

Event-driven scheduling is a sub-type of data-aware scheduling where a DAG is triggered when messages are posted to a message queue. This is useful for scenarios where you want to trigger a DAG based on events that occur outside of Airflow, such as data delivery to an external system or IoT sensor events, and is key to inference execution pipelines.

In this guide you will learn about the concepts used in event-driven scheduling, common usage patterns, and how to implement an example using Amazon SQS and an example using Apache Kafka.

Assumed knowledge

To get the most out of this guide, you should have an existing knowledge of:

Concepts

There are a number of concepts that are important to understand when using event-driven scheduling.

  • Data-aware scheduling: Data-aware or data-driven scheduling refers to all ways you can schedule a DAG based on updates to assets. Updates to assets outside of event-driven scheduling occur by tasks in the same Airflow instance completing successfully, manually through the Airflow UI, or through a call to the Airflow REST API. See Assets and data-aware scheduling in Airflow.
  • Event-driven scheduling: Event-driven scheduling is a sub-type of data-aware scheduling where a DAG is run based on messages posted to a message queue. This message in the queue is triggered by an event that occurs outside of Airflow.
  • Message queue: A message queue is a service that allows you to send and receive messages between different systems. Examples of message queues include Amazon SQS, RabbitMQ, and Apache Kafka. Event-driven scheduling in Airflow 3.0 is supported for Amazon SQS with support for other message queues planned for future releases.
  • Trigger: A trigger is an asynchronous Python function running in the Airflow triggerer component. Triggers that inherit from BaseEventTrigger can be used in AssetWatchers for event-driven scheduling. The trigger is responsible for polling the message queue for new messages, when a new message is found, a TriggerEvent is created. The message is deleted from the queue.
  • AssetWatcher: An AssetWatcher is a class in Airflow that watches one or more triggers for events. When a trigger fires a TriggerEvent, the AssetWatcher updates the asset it is associated with, creating an AssetEvent. The payload of the trigger is attached to the AssetEvent in its extra dictionary.
  • AssetEvent: An Asset is an object in Airflow that represents a concrete or abstract data entity. For example, an asset can be a file, table in a database, or not tied to any specific data. An AssetEvent represents one update to an asset. In the context of event-driven scheduling, the AssetEvent represents one message having been detected in the message queue.

When to use event-driven scheduling

Basic and advanced data-aware scheduling are great for use cases where updates to assets occur within Airflow or can be accomplished via a call to the Airflow REST API. However, there are scenarios where you need a DAG to run based on events in external systems. Two common patterns exist for event-driven scheduling:

  1. Data delivery to an external system: Data is delivered to an external system, such as manually by a domain expert, and a data-ready event is sent to a message queue. The DAG in Airflow is scheduled based on this message event and runs an extract-transform-load (ETL) pipeline that processes the data in the external system.

  2. IoT sensor events: An Internet of Things (IoT) device sends a sensor event to a message queue. A DAG in Airflow is scheduled based on this message event and consumes the message to evaluate the sensor value. If the evaluation determines that an alert is warranted, an alert event is published to another message queue.

Two patterns using event-driven scheduling.

One common use case for event-driven scheduling is inference execution, where the DAG that is triggered involves a call to a machine learning model. Airflow can be used to orchestrate inference execution pipelines of all types, including in Generative AI applications. A key change enabling inference execution in Airflow 3.0 is that a DAG can be triggered with None provided as the logical_date, meaning simultaneous triggering of multiple DAG runs is possible.

Info

Currently, the MessageQueueTrigger works with Amazon SQS and Apache Kafka out-of-the-box with support for other message queues planned for future releases. You can create your own triggers to use with AssetWatchers by inheriting from the BaseEventTrigger class. See the Airflow documentation for more information on supported triggers for event-driven scheduling.

Example: Amazon SQS

This example shows how to configure a DAG to run as soon as a message is posted to an Amazon SQS queue.

  1. Create an Amazon SQS queue. See Amazon Simple Queue Service Documentation for instructions.

  2. Add the Airflow Common Messaging provider and Airflow Amazon provider to your Airflow instance. When using the Astro CLI, you can add the providers to your requirements.txt file:

    apache-airflow-providers-amazon>=9.7.0
    apache-airflow-providers-common-messaging>=1.0.2
    aiobotocore

  3. Set the connection to your Amazon SQS queue in your Airflow instance. Note that the connection needs to include the region_name in the extra field. Replace <your access key>, <your secret key>, and <your region> with your AWS credentials and region. For other authentication options see the Airflow Amazon provider documentation.

    AIRFLOW_CONN_AWS_DEFAULT='{
        "conn_type":"aws",
        "login":"<your access key>",
        "password":"<you secret key>",
        "extra": {
            "region_name":"<your region>",
        }
    }'

    Your AWS user requires at least sqs:ReceiveMessage and sqs:DeleteMessage permissions.

  1. Create a new file in the dags folder of your Airflow project and add the following code. Replace the SQS_QUEUE with the URL to your message queue.

    1from airflow.providers.common.messaging.triggers.msg_queue import MessageQueueTrigger
    2from airflow.sdk import Asset, AssetWatcher, dag, task
    3import os
    4
    5# Define the SQS queue URL
    6SQS_QUEUE = "https://sqs.<region>.amazonaws.com/<acct id>/<queue name>"
    7
    8# Define a trigger that listens to an external message queue (AWS SQS in this case)
    9trigger = MessageQueueTrigger(
    10    aws_conn_id="aws_default",
    11    queue=SQS_QUEUE,
    12    waiter_delay=30,  # delay in seconds between polls
    13)
    14
    15# Define an asset that watches for messages on the queue
    16sqs_queue_asset = Asset(
    17    "sqs_queue_asset", watchers=[AssetWatcher(name="sqs_watcher", trigger=trigger)]
    18)
    19
    20
    21# Schedule the DAG to run when the asset is triggered
    22@dag(schedule=[sqs_queue_asset])
    23def event_driven_dag():
    24    @task
    25    def process_message(**context):
    26        # Extract the triggering asset events from the context
    27        triggering_asset_events = context["triggering_asset_events"]
    28        for event in triggering_asset_events[sqs_queue_asset]:
    29            # Get the message from the TriggerEvent payload
    30            print(
    31                f'Processing message: {event.extra["payload"]["message_batch"][0]["Body"]}'
    32            )
    33
    34    process_message()
    35
    36
    37event_driven_dag()

    This DAG is scheduled to run as soon as the sqs_queue_asset asset is updated. This asset uses one AssetWatcher with the name sqs_watcher that watches one MessageQueueTrigger. This trigger is polling for new messages in the provided SQS queue.

    The process_message task gets the triggering asset events from the Airflow context and prints the message body from the triggering message. The process_message task is a placeholder for your own task that processes the message.

  2. Create a new message in the SQS queue. It will trigger the DAG to run and the process_message task will print the message body.

Example: Apache Kafka

To use Apache Kafka as the message queue for event-driven scheduling, follow these steps:

  1. Create an Apache Kafka topic. See the Apache Kafka documentation for instructions.

  2. Add the Airflow Common Messaging provider and Airflow Apache Kafka provider to your Airflow instance. When using the Astro CLI, you can add the providers to your requirements.txt file:

    apache-airflow-providers-apache-kafka>=1.9.0
    apache-airflow-providers-common-messaging>=1.0.2

  3. Set the connection to your Apache Kafka topic in your Airflow instance. Below is an example connection json, depending on your Kafka setup you may need to provide additional values in the extra dictionary.

    AIRFLOW_CONN_KAFKA_DEFAULT='{
        "conn_type": "general",
        "extra": {
            "bootstrap.servers": "<your host>:<your port>",
            "group.id": "<your group id>",
            "security.protocol": "<your security protocol>",
            "enable.auto.commit": false,
            "auto.offset.reset": "beginning"
        }
    }'

  4. Create a new file in the dags folder of your Airflow project and add the following code. Replace the KAFKA_TOPIC with the name of your Kafka topic.

    1import json
    2
    3from airflow.providers.common.messaging.triggers.msg_queue import MessageQueueTrigger
    4from airflow.providers.standard.operators.empty import EmptyOperator
    5from airflow.sdk import dag, Asset, AssetWatcher, task
    6
    7# Define the Kafka queue URL
    8# Replace <your_kafka_host>, <port>, and <your_topic> with your Kafka
    9KAFKA_QUEUE = "kafka://<your_kafka_host>:<port>/<your_topic>"
    10
    11
    12# Define a function to apply when a message is received
    13# This function will be called with the message as an argument
    14def apply_function(*args, **kwargs):
    15    message = args[-1]
    16    val = json.loads(message.value())
    17    print(f"Value in message is {val}")
    18    return val
    19
    20
    21# Define a trigger that listens to an external message queue (Kafka in this case)
    22trigger = MessageQueueTrigger(
    23    queue=KAFKA_QUEUE,
    24    apply_function="dags.event_driven_dag.apply_function",
    25)
    26
    27# Define an asset that watches for messages on the Kafka topic
    28kafka_topic_asset = Asset(
    29    "kafka_topic_asset", watchers=[AssetWatcher(name="kafka_watcher", trigger=trigger)]
    30)
    31
    32
    33@dag(schedule=[kafka_topic_asset])
    34def event_driven_dag():
    35    @task
    36    def process_message(**context):
    37        # Extract the triggering asset events from the context
    38        triggering_asset_events = context["triggering_asset_events"]
    39        for event in triggering_asset_events[kafka_topic_asset]:
    40            # Get the message from the TriggerEvent
    41            print(f"Processing message: {event}")
    42
    43    process_message()
    44
    45
    46event_driven_dag()

    This DAG is scheduled to run as soon as the kafka_topic_asset asset is updated. This asset uses one AssetWatcher with the name kafka_watcher that watches one MessageQueueTrigger. This trigger is polling for new messages in the provided Kafka topic.

    The process_message task gets the triggering asset events from the Airflow context and prints the event information from the trigger event that includes the message. The process_message task is a placeholder for your own task that processes the message.

  5. Create a new message in the Kafka topic. It will trigger the DAG to run and the process_message task will print the event information.