cmarteepants commented on PR #65103:
URL: https://github.com/apache/airflow/pull/65103#issuecomment-4284138414
+1 on the direction of making watcher-side triggers asset-aware. AIP 103
doesn't really land without something like this. Before we commit to the shape
though, I want to raise a second case that's been hindering event-driven
scheduling adoption, because I think it's worth deciding up front whether we
scope it into this work or handle it separately.
Today's model gives every asset its own watcher, and every watcher its own
trigger. With `MessageQueueTrigger`, that means each asset polls the queue
independently. In practice, a single SQS queue or Kafka topic usually caries
events for many logical assets, so users with 25 assets on one queue end up
with 25 trigger instances polling the same queue and filtering out messages
they don't care about. Which means more polling than they actually need +
coordination concerns (visibility timeouts, independent consumer offsets, etc).
That's more polling than they need, plus coordination concerns (visibility
timeouts, independent consumer offsets, etc.) they didn't sign up for. Once
customers hit this, they usually drop off from using the feature altogether and
move the fan-out logic into a Lambda or other serverless compute that calls
Airflow's REST API instead.
So there's really two cases this interface needs to support:
1. **Watermark / per resource** (what this PR solves): One AssetWatcher <-->
One Asset, trigger reads that asset's state via AIP-103.
2. **Fan-In** (not solved here): One watcher consumes a shared queue once
and emits events for whichever asset each message pertains to.
At the watcher-declaration level, these could live under one shape pretty
naturally. Maybe something like this?
```
from airflow.sdk import Asset, AssetWatcher
orders = Asset("orders")
refunds = Asset("refunds")
shipments = Asset("shipments")
users_table = Asset("users_table")
# Fan-in: one watcher, one queue, many assets
orders_queue_watcher = AssetWatcher(
name="orders_queue",
assets=[orders, refunds, shipments],
trigger=SQSFanInTrigger(
queue_url="https://sqs.us-east-1.amazonaws.com/.../orders-events";,
routing={
"order_created": orders,
"refund_issued": refunds,
"shipment_dispatched": shipments,
},
),
)
# Watermark: one watcher, one asset (uses AIP-103 state)
users_watermark_watcher = AssetWatcher(
name="users_watermark",
assets=[users_table],
trigger=TableWatermarkTrigger(table="users"),
)
```
The watcher signature (`asset` vs `assets=[...]`) is probably the smallest
part. I think the change fan-in would need is in the trigger --> scheduler
contract. The trigger needs a way to say "this even is for asset X" so the
scheduler can route it to the right asset event, which is something this PR
doesn't touch since it's 1:1.
So the real question is: Do we scope the fan-in case now or later? My
concern with doing this later is that we are going to be formalizing the 1:1
contract. Evolving `TriggerEvent` and the watcher-assets relationship after the
fact will be meaningfully harder, and the strong coupling today is already a
pretty significant barrier to adoption. Happy to spin up a separate discussion
either way.
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
