You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@airflow.apache.org by "jroachgolf84 (via GitHub)" <gi...@apache.org> on 2025/05/08 19:02:27 UTC
[I] Patterns for Building Custom Asset Watchers [airflow]
jroachgolf84 opened a new issue, #50369:
URL: https://github.com/apache/airflow/issues/50369
### 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?
- [x] Yes I am willing to submit a PR!
### Code of Conduct
- [x] I agree to follow this project's [Code of Conduct](https://github.com/apache/airflow/blob/main/CODE_OF_CONDUCT.md)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [I] Patterns for Building Custom Asset Watchers [airflow]
Posted by "tirkarthi (via GitHub)" <gi...@apache.org>.
tirkarthi commented on issue #50369:
URL: https://github.com/apache/airflow/issues/50369#issuecomment-2864076518
cc: @vincbeck
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [I] Patterns for Building Custom Asset Watchers [airflow]
Posted by "vincbeck (via GitHub)" <gi...@apache.org>.
vincbeck commented on issue #50369:
URL: https://github.com/apache/airflow/issues/50369#issuecomment-2872323667
> 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.
It is definitely the former. The idea is, any trigger extending `BaseEventTrigger` can be used to schedule DAGs. `AssetWatcher` is only a wrapper on top of the trigger, essentially it allows to name the trigger. This will be useful for #48017. The more triggers extending `BaseEventTrigger` there are available, the more options users will have to schedule their DAGs based on events.
> 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.
`AssetWatcher` does not pass any information to downstream tasks, again this is just a thin wrapper on top of the trigger. On the other side, the trigger can pass information to the downstream tasks using the event. Using `triggering_asset_events`, you can fetch the events that triggered the DAG, so any information you pass when sending the event in the trigger, can be fetch in tasks.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [I] Patterns for Building Custom Asset Watchers [airflow]
Posted by "vincbeck (via GitHub)" <gi...@apache.org>.
vincbeck commented on issue #50369:
URL: https://github.com/apache/airflow/issues/50369#issuecomment-2876481717
Done ;)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [I] Patterns for Building Custom Asset Watchers [airflow]
Posted by "jroachgolf84 (via GitHub)" <gi...@apache.org>.
jroachgolf84 commented on issue #50369:
URL: https://github.com/apache/airflow/issues/50369#issuecomment-2864011667
cc: @cmarteepants
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [I] Patterns for Building Custom Asset Watchers [airflow]
Posted by "jroachgolf84 (via GitHub)" <gi...@apache.org>.
jroachgolf84 commented on issue #50369:
URL: https://github.com/apache/airflow/issues/50369#issuecomment-2876474512
cc: @RNHTTR, can you assign this to me?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org