Description

Less of a feature request and more of a conversation around how we'll start to build custom Asset Watchers.

Is the general pattern going to be to build a Trigger, and leverage the out-of-the-box airflow.sdk.AssetWatcher when defining an Asset? Or instead, would a user/contributor define a custom Asset Watcher that has a baked-in Trigger.

How will we pass information from an AssetWatcher object to downstream Tasks in a DAG? Will this always be via context['triggering_asset_events'] ...? The most obvious example would be a file lands in an object store, and we want to process it.

Here's an example that's available in Astronomer's documentation, using SQS.

# Define an asset that watches for messages on the queue
sqs_queue_asset = Asset(
    "sqs_queue_asset", watchers=[AssetWatcher(name="sqs_watcher", trigger=trigger)]
)


# Schedule the DAG to run when the asset is triggered
@dag(schedule=[sqs_queue_asset])
def event_driven_dag():
    @task
    def process_message(**context):
        # Extract the triggering asset events from the context
        triggering_asset_events = context["triggering_asset_events"]
        for event in triggering_asset_events[sqs_queue_asset]:
            # Get the message from the TriggerEvent payload
            print(
                f"Processing message: {event.extra["payload"]["message_batch"][0]["Body"]}"
            )

    process_message()


event_driven_dag()

Use case/motivation

Event-driven (Asset-aware) scheduling is (IMO) going to be one of the most powerful features that Airflow users will leverage. I'm hoping to lay the groundwork for this as both users create custom Asset Watchers and contributors add to providers.

Related issues

No response

Are you willing to submit a PR?

• Yes I am willing to submit a PR!

Code of Conduct

• I agree to follow this project's Code of Conduct