You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@airflow.apache.org by "renat-sagut (via GitHub)" <gi...@apache.org> on 2026/06/13 21:32:46 UTC
[PR] batch asset events [airflow]
renat-sagut opened a new pull request, #68517:
URL: https://github.com/apache/airflow/pull/68517
<!-- SPDX-License-Identifier: Apache-2.0
https://www.apache.org/licenses/LICENSE-2.0 -->
<!--
Thank you for contributing!
Please provide above a brief description of the changes made in this pull request.
Write a good git commit message following this guide: http://chris.beams.io/posts/git-commit/
Please make sure that your code changes are covered with tests.
And in case of new features or big changes remember to adjust the documentation.
For user-facing UI changes, please attach before/after screenshots (or a short
screen recording) so reviewers can assess the visual impact.
Feel free to ping (in general) for the review if you do not see reaction for a few days
(72 Hours is the minimum reaction time you can expect from volunteers) - we sometimes miss notifications.
In case of an existing issue, reference it using one of the following:
* closes: #ISSUE
* related: #ISSUE
-->
---
##### Was generative AI tooling used to co-author this PR?
<!--
If generative AI tooling has been used in the process of authoring this PR, please
change below checkbox to `[X]` followed by the name of the tool, uncomment the "Generated-by".
-->
- [ ] Yes (please specify the tool below)
<!--
Generated-by: [Tool Name] following [the guidelines](https://github.com/apache/airflow/blob/main/contributing-docs/05_pull_requests.rst#gen-ai-assisted-contributions)
-->
---
* Read the **[Pull Request Guidelines](https://github.com/apache/airflow/blob/main/contributing-docs/05_pull_requests.rst#pull-request-guidelines)** for more information. Note: commit author/co-author name and email in commits become permanently public when merged.
* For fundamental code changes, an Airflow Improvement Proposal ([AIP](https://cwiki.apache.org/confluence/display/AIRFLOW/Airflow+Improvement+Proposals)) is needed.
* When adding dependency, check compliance with the [ASF 3rd Party License Policy](https://www.apache.org/legal/resolved.html#category-x).
* For significant user-facing changes create newsfragment: `{pr_number}.significant.rst`, in [airflow-core/newsfragments](https://github.com/apache/airflow/tree/main/airflow-core/newsfragments). You can add this file in a follow-up commit after the PR is created so you know the PR number.
--
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: [PR] Make asset event batching opt-in [airflow]
Posted by "cmarteepants (via GitHub)" <gi...@apache.org>.
cmarteepants commented on PR #68517:
URL: https://github.com/apache/airflow/pull/68517#issuecomment-6045372849
>I think we should still have a newsfragment since this changes the default behavior (batch_asset_events = False). Either change the default to match the current behavior, or we should tell the user about the change.
@ephraimbuddy just confirming we're going with the newsfragment option right?
--
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: [PR] Make asset event batching opt-in [airflow]
Posted by "ephraimbuddy (via GitHub)" <gi...@apache.org>.
ephraimbuddy commented on PR #68517:
URL: https://github.com/apache/airflow/pull/68517#issuecomment-6020830384
Pushed three fixes on top of this PR, following the dev-list agreement that batching should be opt-in:
- **Opt-in batching:** `batch_asset_events` now defaults to the new `[scheduler] batch_asset_events` option (false), and setting it to true restores the old behaviour deployment-wide.
- **AND/rollup exception:** `a & b` conditions and partition rollups can't run one event at a time, so they still default to batching when the argument isn't passed, and only an explicit `False` for them is rejected, which keeps existing Dags parsing.
- **Bounded run creation:** with batching off, the events picked up per loop are capped by `max_dagruns_to_create_per_loop` and each Dag's `max_active_runs` headroom, and the rest stay queued.
- **Partitioned pending rows:** the UI detail endpoint now takes a `partitioned_dag_run_id` (falling back to the oldest row), and enabling batching merges pending rows for the same key into the newest one without losing events.
--
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: [PR] Add batch_asset_events parameter to asset-triggered timetables [airflow]
Posted by "cmarteepants (via GitHub)" <gi...@apache.org>.
cmarteepants commented on PR #68517:
URL: https://github.com/apache/airflow/pull/68517#issuecomment-5702054976
Thanks for picking this up @renat-sagut ! Before this lands, I started a dev-list discussion on what the default should be, advocating that we should NOT be batching by default, even if it would technically be a breaking change. The thread came out consistently in favour of **not** batching by default (1 event -> 1 DagRun):
- **Jake McGrath**: sees a real use case for batching, but leans toward not batching asset events by default.
- **Jarek Potiuk**: +1 on not batching, and would deprecate/remove auto-batching entirely — the batch size is unpredictable and uncontrollable, so unrelated load on a deployment can silently change a Dag's behavior.
- **Philippe Gagnon**: agrees the behavior is counter-intuitive.
- **Kaxil Naik**: had assumed it was a bug because it's hard to reason about, and thinks we should change it.
Defaulting to today's behaviour looks like the safer choice, but batch size already varies with scheduler cadence and parallelism, so nobody can be relying on a specific value, and it's very misunderstood. The cost of unbatching is an occasional extra DagRun for a few users while the cost of keeping it is everyone else either has to deal with extra events and break their pipelines if not accounted for (which happens a lot since it's very misunderstood as to how this works).
Would you be willing to flip the default so that batching is opt-in rather than opt-out?
Dev list email chain: https://lists.apache.org/thread/jz98dor8h02mjoxddtdp1cvml9d8yxcr
--
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: [PR] batch asset events [airflow]
Posted by "vatsrahul1001 (via GitHub)" <gi...@apache.org>.
vatsrahul1001 commented on PR #68517:
URL: https://github.com/apache/airflow/pull/68517#issuecomment-4767464184
@renat-sagut looks like we have a conflict can you resolve
--
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: [PR] batch asset events [airflow]
Posted by "renat-sagut (via GitHub)" <gi...@apache.org>.
renat-sagut commented on code in PR #68517:
URL: https://github.com/apache/airflow/pull/68517#discussion_r3650095406
##########
airflow-core/src/airflow/assets/manager.py:
##########
@@ -666,6 +667,7 @@ def _get_or_create_apdr(
rollup_fingerprint: dict,
asset_id: int,
session: Session,
+ allow_reuse: bool = True,
Review Comment:
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: [PR] batch asset events [airflow]
Posted by "uranusjr (via GitHub)" <gi...@apache.org>.
uranusjr commented on code in PR #68517:
URL: https://github.com/apache/airflow/pull/68517#discussion_r3744714466
##########
airflow-core/docs/authoring-and-scheduling/asset-scheduling.rst:
##########
@@ -419,3 +419,29 @@ AssetTimetable Integration
You can schedule Dags based on both asset events and time-based schedules using ``AssetOrTimeSchedule``. This allows you to create workflows when a Dag needs both to be triggered by data updates and run periodically according to a fixed timetable.
For more detailed information on ``AssetOrTimeSchedule``, refer to the corresponding section in :ref:`AssetOrTimeSchedule <asset-timetable-section>`.
+
+
+Controlling DagRun creation per asset event
+---------------------------------------------
+
+.. versionadded:: 3.3.1
Review Comment:
This needs to be 3.4.0, we’re not going to backport this.
--
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: [PR] Make asset event batching opt-in [airflow]
Posted by "uranusjr (via GitHub)" <gi...@apache.org>.
uranusjr commented on code in PR #68517:
URL: https://github.com/apache/airflow/pull/68517#discussion_r4203988696
##########
airflow-core/src/airflow/timetables/base.py:
##########
@@ -228,6 +228,9 @@ class Timetable(Protocol):
asset_condition: SerializedAssetBase = _NullAsset()
"""The asset condition that triggers a DAG using this timetable."""
+ batch_asset_events: bool = False
Review Comment:
Since this is always changed by configuration, it probably should have a default.
--
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: [PR] Make asset event batching opt-in [airflow]
Posted by "ephraimbuddy (via GitHub)" <gi...@apache.org>.
ephraimbuddy commented on code in PR #68517:
URL: https://github.com/apache/airflow/pull/68517#discussion_r4204262630
##########
airflow-core/src/airflow/timetables/base.py:
##########
@@ -228,6 +228,9 @@ class Timetable(Protocol):
asset_condition: SerializedAssetBase = _NullAsset()
"""The asset condition that triggers a DAG using this timetable."""
+ batch_asset_events: bool = False
Review Comment:
Do you mean configuration default here?
--
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: [PR] batch asset events [airflow]
Posted by "renat-sagut (via GitHub)" <gi...@apache.org>.
renat-sagut commented on PR #68517:
URL: https://github.com/apache/airflow/pull/68517#issuecomment-4702008716
> Early feedback while CI is running: Can you add some documentation with a very simple example to airflow-core/docs/authoring-and-scheduling/asset-scheduling.rst?
>
> Note for reference, this is related to Slack discussion in https://apache-airflow.slack.com/archives/C06K9Q5G2UA/p1778258939480149
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: [PR] Make asset event batching opt-in [airflow]
Posted by "uranusjr (via GitHub)" <gi...@apache.org>.
uranusjr commented on code in PR #68517:
URL: https://github.com/apache/airflow/pull/68517#discussion_r4204109926
##########
airflow-core/src/airflow/timetables/assets.py:
##########
@@ -47,8 +47,9 @@ def __init__(
*,
timetable: Timetable,
assets: Collection[SerializedAsset] | SerializedAssetBase,
+ batch_asset_events: bool | None = None,
Review Comment:
Instead of `bool | None`, maybe this class should use attrs or dataclass factory to keep the argument typed as simply `bool`. This would also remove the need for coercing None manually inside `__init__` (which can be a bit difficult to maintain).
--
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: [PR] batch asset events [airflow]
Posted by "uranusjr (via GitHub)" <gi...@apache.org>.
uranusjr commented on code in PR #68517:
URL: https://github.com/apache/airflow/pull/68517#discussion_r3457052535
##########
airflow-core/src/airflow/jobs/scheduler_job_runner.py:
##########
@@ -2572,27 +2572,34 @@ def _create_dag_runs_asset_triggered(
)
)
- dag_run = dag.create_dagrun(
- run_id=DagRun.generate_run_id(
- run_type=DagRunType.ASSET_TRIGGERED, logical_date=None, run_after=triggered_date
- ),
- logical_date=None,
- data_interval=None,
- run_after=triggered_date,
- run_type=DagRunType.ASSET_TRIGGERED,
- triggered_by=DagRunTriggeredByType.ASSET,
- state=DagRunState.QUEUED,
- creating_job_id=self.job.id,
- session=session,
- )
- stats.incr("asset.triggered_dagruns")
- dag_run.consumed_asset_events.extend(asset_events)
- self.log.info(
- "Created asset-triggered DagRun for '%s': run_id=%s, consumed %d asset events",
- dag.dag_id,
- dag_run.run_id,
- len(asset_events),
- )
+ # Build the list of (run_after, events) to process: one entry per DagRun to create.
+ if dag.timetable.batch_asset_events:
+ event_runs = [(triggered_date, asset_events)]
+ else:
+ event_runs = [(timezone.coerce_datetime(ev.timestamp), [ev]) for ev in asset_events]
Review Comment:
```suggestion
event_runs = ((timezone.coerce_datetime(ev.timestamp), [ev]) for ev in asset_events)
```
This does not need to be a list since it’s simply consumed directly.
(Typing tweaks may be needed to satisfy the type checker.)
--
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: [PR] Make asset event batching opt-in [airflow]
Posted by "uranusjr (via GitHub)" <gi...@apache.org>.
uranusjr commented on code in PR #68517:
URL: https://github.com/apache/airflow/pull/68517#discussion_r4232381268
##########
airflow-core/src/airflow/timetables/base.py:
##########
@@ -228,6 +228,9 @@ class Timetable(Protocol):
asset_condition: SerializedAssetBase = _NullAsset()
"""The asset condition that triggers a DAG using this timetable."""
+ batch_asset_events: bool = False
Review Comment:
I think this is fine now.
--
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: [PR] batch asset events [airflow]
Posted by "potiuk (via GitHub)" <gi...@apache.org>.
potiuk commented on PR #68517:
URL: https://github.com/apache/airflow/pull/68517#issuecomment-4947259003
@renat-sagut I've removed the `ready for maintainer review` label because the next step here is yours: unresolved review thread(s) from `@uranusjr` awaiting your reply. It'll go back into the maintainer queue automatically once that's done — no need to re-add the label by hand. Next move: address the unresolved threads and reply to `@uranusjr`. There is no rush.
---
_Note: This comment was drafted by an AI-assisted triage tool and may contain mistakes. If you have questions, an Apache Airflow maintainer — a real person — is happy to help. We use this [two-stage triage process](https://github.com/apache/airflow/blob/main/contributing-docs/25_maintainer_pr_triage.md#why-the-first-pass-is-automated) so that our maintainers' limited time is spent where it matters most: the conversation with you._
--
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: [PR] batch asset events [airflow]
Posted by "renat-sagut (via GitHub)" <gi...@apache.org>.
renat-sagut commented on PR #68517:
URL: https://github.com/apache/airflow/pull/68517#issuecomment-5078310036
> @renat-sagut looks like we have a conflict can you resolve
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: [PR] Make asset event batching opt-in [airflow]
Posted by "uranusjr (via GitHub)" <gi...@apache.org>.
uranusjr commented on code in PR #68517:
URL: https://github.com/apache/airflow/pull/68517#discussion_r4204069269
##########
airflow-core/src/airflow/jobs/scheduler_job_runner.py:
##########
@@ -2931,41 +2958,54 @@ def _create_dag_runs_asset_triggered(
len(queued_adrqs),
triggered_date,
)
- dag_run = dag.create_dagrun(
- run_id=DagRun.generate_run_id(
- run_type=DagRunType.ASSET_TRIGGERED, logical_date=None, run_after=triggered_date
- ),
- logical_date=None,
- data_interval=None,
- run_after=triggered_date,
- run_type=DagRunType.ASSET_TRIGGERED,
- triggered_by=DagRunTriggeredByType.ASSET,
- state=DagRunState.QUEUED,
- creating_job_id=self.job.id,
- session=session,
- )
+ if dag.timetable.batch_asset_events:
+ event_runs = iter([(triggered_date, asset_events)])
+ else:
+ event_runs = ((timezone.coerce_datetime(ev.timestamp), [ev]) for ev in asset_events)
Review Comment:
The scheduler now branches on `timetable.batch_asset_events` in several places. Grouping events into runs should be the timetable's decision. Maybe we should have a method on the timetable
```python
def group_asset_events(
self,
events,
triggered_date,
) -> Iterable[tuple[datetime, list[AssetEvent]]]
```
for the scheduler to use?
--
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: [PR] Make asset event batching opt-in [airflow]
Posted by "ephraimbuddy (via GitHub)" <gi...@apache.org>.
ephraimbuddy commented on PR #68517:
URL: https://github.com/apache/airflow/pull/68517#issuecomment-6077468226
> > I think we should still have a newsfragment since this changes the default behavior (batch_asset_events = False). Either change the default to match the current behavior, or we should tell the user about the change.
>
> @ephraimbuddy just confirming we're going with the newsfragment option right?
Yes. I added the newsfragment
--
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: [PR] Make asset event batching opt-in [airflow]
Posted by "uranusjr (via GitHub)" <gi...@apache.org>.
uranusjr commented on code in PR #68517:
URL: https://github.com/apache/airflow/pull/68517#discussion_r4232394711
##########
airflow-core/docs/authoring-and-scheduling/asset-scheduling.rst:
##########
Review Comment:
We might want to mention the catchup behavior here too?
--
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: [PR] batch asset events [airflow]
Posted by "renat-sagut (via GitHub)" <gi...@apache.org>.
renat-sagut commented on code in PR #68517:
URL: https://github.com/apache/airflow/pull/68517#discussion_r3650097245
##########
airflow-core/src/airflow/jobs/scheduler_job_runner.py:
##########
@@ -2572,27 +2572,34 @@ def _create_dag_runs_asset_triggered(
)
)
- dag_run = dag.create_dagrun(
- run_id=DagRun.generate_run_id(
- run_type=DagRunType.ASSET_TRIGGERED, logical_date=None, run_after=triggered_date
- ),
- logical_date=None,
- data_interval=None,
- run_after=triggered_date,
- run_type=DagRunType.ASSET_TRIGGERED,
- triggered_by=DagRunTriggeredByType.ASSET,
- state=DagRunState.QUEUED,
- creating_job_id=self.job.id,
- session=session,
- )
- stats.incr("asset.triggered_dagruns")
- dag_run.consumed_asset_events.extend(asset_events)
- self.log.info(
- "Created asset-triggered DagRun for '%s': run_id=%s, consumed %d asset events",
- dag.dag_id,
- dag_run.run_id,
- len(asset_events),
- )
+ # Build the list of (run_after, events) to process: one entry per DagRun to create.
+ if dag.timetable.batch_asset_events:
+ event_runs = [(triggered_date, asset_events)]
+ else:
+ event_runs = [(timezone.coerce_datetime(ev.timestamp), [ev]) for ev in asset_events]
Review Comment:
each element is consumed separately when batch_asset_events is False
--
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: [PR] batch asset events [airflow]
Posted by "uranusjr (via GitHub)" <gi...@apache.org>.
uranusjr commented on PR #68517:
URL: https://github.com/apache/airflow/pull/68517#issuecomment-5232951360
We’ll need to change the docstring on AssetPartitionDagRun. It currently says
```
We should not allow more than one row with the same target_dag_id /
partition_key where created_dag_run_id is null, and this is what the
`_lock_asset_model` mutex control is for.
```
I believe this is no longer true with `allow_reuse=False`?
--
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: [PR] Add batch_asset_events parameter to asset-triggered timetables [airflow]
Posted by "jroachgolf84 (via GitHub)" <gi...@apache.org>.
jroachgolf84 commented on PR #68517:
URL: https://github.com/apache/airflow/pull/68517#issuecomment-5701882997
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: [PR] Make asset event batching opt-in [airflow]
Posted by "uranusjr (via GitHub)" <gi...@apache.org>.
uranusjr commented on code in PR #68517:
URL: https://github.com/apache/airflow/pull/68517#discussion_r4204125760
##########
airflow-core/src/airflow/assets/manager.py:
##########
@@ -749,7 +757,7 @@ def _get_or_create_apdr(
.order_by(AssetPartitionDagRun.id.desc())
.limit(1)
)
- if latest_apdr and latest_apdr.created_dag_run_id is None:
+ if latest_apdr and latest_apdr.created_dag_run_id is None and allow_reuse:
Review Comment:
When `allow_reuse=False`, we’d still run the latest_apdr select under the lock, and then ignore the result. Skip the query instead of adding the check here.
--
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: [PR] Make asset event batching opt-in [airflow]
Posted by "uranusjr (via GitHub)" <gi...@apache.org>.
uranusjr commented on PR #68517:
URL: https://github.com/apache/airflow/pull/68517#issuecomment-6032707483
I think we should still have a newsfragment since this changes the default behavior (`batch_asset_events = False`). Either change the default to match the current behavior, or we should tell the user about the change.
--
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: [PR] batch asset events [airflow]
Posted by "jscheffl (via GitHub)" <gi...@apache.org>.
jscheffl commented on PR #68517:
URL: https://github.com/apache/airflow/pull/68517#issuecomment-4701917174
Early feedback while CI is running: Can you add some documentation with a very simple example to airflow-core/docs/authoring-and-scheduling/asset-scheduling.rst?
Note for reference, this is related to Slack discussion in https://apache-airflow.slack.com/archives/C06K9Q5G2UA/p1778258939480149
--
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: [PR] batch asset events [airflow]
Posted by "uranusjr (via GitHub)" <gi...@apache.org>.
uranusjr commented on code in PR #68517:
URL: https://github.com/apache/airflow/pull/68517#discussion_r3457025617
##########
airflow-core/src/airflow/assets/manager.py:
##########
@@ -666,6 +667,7 @@ def _get_or_create_apdr(
rollup_fingerprint: dict,
asset_id: int,
session: Session,
+ allow_reuse: bool = True,
Review Comment:
Is the default needed? This is a private function so we can change all the call sites instead. A default should be avoided unless necessarily to prevent calls missing an explicit flag silently do the wrong thing.
Also, the `session` argument should generally be put last.
--
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: [PR] batch asset events [airflow]
Posted by "uranusjr (via GitHub)" <gi...@apache.org>.
uranusjr commented on code in PR #68517:
URL: https://github.com/apache/airflow/pull/68517#discussion_r3744727847
##########
airflow-core/src/airflow/timetables/simple.py:
##########
@@ -230,14 +233,20 @@ def __init__(self, assets: Collection[SerializedAsset] | SerializedAssetBase) ->
def deserialize(cls, data: dict[str, Any]) -> Timetable:
from airflow.serialization.decoders import decode_asset_like
- return cls(decode_asset_like(data["asset_condition"]))
+ return cls(
+ decode_asset_like(data["asset_condition"]),
+ batch_asset_events=data.get("batch_asset_events", True),
+ )
@property
def summary(self) -> str:
return "Asset"
def serialize(self) -> dict[str, Any]:
- return {"asset_condition": encode_asset_like(self.asset_condition)}
+ return {
+ "asset_condition": encode_asset_like(self.asset_condition),
+ "batch_asset_events": self.batch_asset_events,
+ } # serialize batch_asset_events flag for the scheduler
Review Comment:
Unnecessary comment
--
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: [PR] Make asset event batching opt-in [airflow]
Posted by "uranusjr (via GitHub)" <gi...@apache.org>.
uranusjr commented on code in PR #68517:
URL: https://github.com/apache/airflow/pull/68517#discussion_r4204119172
##########
airflow-core/src/airflow/api_fastapi/core_api/routes/ui/partitioned_dag_runs.py:
##########
@@ -390,11 +391,11 @@ def get_pending_partitioned_dag_run(
AssetPartitionDagRun.partition_key == partition_key,
AssetPartitionDagRun.created_dag_run_id.is_(None),
)
- # Duplicate pending rows for the same (dag_id, partition_key) can exist
- # after a crash; mirror _get_or_create_apdr and work on the latest one.
- .order_by(AssetPartitionDagRun.id.desc())
- .limit(1)
- ).first()
+ .order_by(AssetPartitionDagRun.created_at, AssetPartitionDagRun.id)
Review Comment:
This should be `id.desc()`?
--
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: [PR] Make asset event batching opt-in [airflow]
Posted by "vatsrahul1001 (via GitHub)" <gi...@apache.org>.
vatsrahul1001 merged PR #68517:
URL: https://github.com/apache/airflow/pull/68517
--
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: [PR] batch asset events [airflow]
Posted by "uranusjr (via GitHub)" <gi...@apache.org>.
uranusjr commented on code in PR #68517:
URL: https://github.com/apache/airflow/pull/68517#discussion_r3457055634
##########
task-sdk/src/airflow/sdk/definitions/timetables/assets.py:
##########
@@ -43,13 +43,13 @@ class AssetTriggeredTimetable(BaseTimetable):
"""
asset_condition: BaseAsset = attrs.field(alias="assets")
+ batch_asset_events: bool = True
@attrs.define
class PartitionedAssetTimetable(AssetTriggeredTimetable):
"""Asset-driven timetable that listens for partitioned assets."""
- asset_condition: BaseAsset = attrs.field(alias="assets")
Review Comment:
Why is this removed?
--
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: [PR] batch asset events [airflow]
Posted by "renat-sagut (via GitHub)" <gi...@apache.org>.
renat-sagut commented on code in PR #68517:
URL: https://github.com/apache/airflow/pull/68517#discussion_r3650098771
##########
task-sdk/src/airflow/sdk/definitions/timetables/assets.py:
##########
@@ -43,13 +43,13 @@ class AssetTriggeredTimetable(BaseTimetable):
"""
asset_condition: BaseAsset = attrs.field(alias="assets")
+ batch_asset_events: bool = True
@attrs.define
class PartitionedAssetTimetable(AssetTriggeredTimetable):
"""Asset-driven timetable that listens for partitioned assets."""
- asset_condition: BaseAsset = attrs.field(alias="assets")
Review Comment:
it is inherited from the parent class
https://github.com/apache/airflow/blob/0378a727e5fcbf63968331944ef7130308039686/task-sdk/src/airflow/sdk/definitions/timetables/assets.py#L45
--
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: [PR] batch asset events [airflow]
Posted by "jscheffl (via GitHub)" <gi...@apache.org>.
jscheffl commented on code in PR #68517:
URL: https://github.com/apache/airflow/pull/68517#discussion_r3410017406
##########
airflow-core/docs/authoring-and-scheduling/asset-scheduling.rst:
##########
@@ -419,3 +419,27 @@ AssetTimetable Integration
You can schedule Dags based on both asset events and time-based schedules using ``AssetOrTimeSchedule``. This allows you to create workflows when a Dag needs both to be triggered by data updates and run periodically according to a fixed timetable.
For more detailed information on ``AssetOrTimeSchedule``, refer to the corresponding section in :ref:`AssetOrTimeSchedule <asset-timetable-section>`.
+
+
+Controlling DagRun creation per asset event
+---------------------------------------------
+
+By default, when multiple asset events arrive for the same Dag between
Review Comment:
nit: Can you add a marker that this is added in 3.3.0?
```suggestion
.. versionadded:: 3.3.0
By default, when multiple asset events arrive for the same Dag between
```
--
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: [PR] Make asset event batching opt-in [airflow]
Posted by "uranusjr (via GitHub)" <gi...@apache.org>.
uranusjr commented on code in PR #68517:
URL: https://github.com/apache/airflow/pull/68517#discussion_r4232391065
##########
airflow-core/src/airflow/timetables/simple.py:
##########
@@ -282,21 +320,28 @@ class PartitionedAssetTimetable(AssetTriggeredTimetable):
def summary(self) -> str:
return "Partitioned Asset"
- def __init__(
- self,
- *,
- assets: SerializedAssetBase,
- partition_mapper_config: dict[SerializedAssetBase, PartitionMapper] | None = None,
- default_partition_mapper: PartitionMapper = DEFAULT_PARTITION_MAPPER,
- ) -> None:
- super().__init__(assets=assets)
- self.partition_mapper_config = partition_mapper_config or {}
- self.default_partition_mapper = default_partition_mapper
-
- self._name_to_partition_mapper: dict[str, PartitionMapper] = {}
- self._uri_to_partition_mapper: dict[str, PartitionMapper] = {}
+ asset_condition: SerializedAssetBase = attrs.field(alias="assets", converter=_coerce_asset_condition)
+ partition_mapper_config: dict[SerializedAssetBase, PartitionMapper] = attrs.field(
+ factory=dict, converter=_coerce_partition_mapper_config
Review Comment:
There’s `attrs.converters.default_if_none`, we don’t need a custom converter here.
--
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: [PR] Make asset event batching opt-in [airflow]
Posted by "uranusjr (via GitHub)" <gi...@apache.org>.
uranusjr commented on code in PR #68517:
URL: https://github.com/apache/airflow/pull/68517#discussion_r4204080677
##########
airflow-core/src/airflow/timetables/simple.py:
##########
@@ -219,26 +221,58 @@ class AssetTriggeredTimetable(_TrivialTimetable):
description: str = "Triggered by assets"
asset_triggered = True
- def __init__(self, assets: Collection[SerializedAsset] | SerializedAssetBase) -> None:
+ def __init__(
+ self,
+ assets: Collection[SerializedAsset] | SerializedAssetBase,
+ *,
+ batch_asset_events: bool | None = None,
+ ) -> None:
super().__init__()
# Compatibility: Handle SDK assets if needed so this class works in dag files.
if isinstance(assets, SerializedAssetBase | BaseAsset):
self.asset_condition = ensure_serialized_asset(assets)
else:
self.asset_condition = SerializedAssetAll([ensure_serialized_asset(a) for a in assets])
+ if batch_asset_events is None:
+ batch_asset_events = (
+ conf.getboolean("scheduler", "batch_asset_events", fallback=False)
+ or self.get_batching_requirement() is not None
+ )
+ self.batch_asset_events = batch_asset_events
@classmethod
def deserialize(cls, data: dict[str, Any]) -> Timetable:
from airflow.serialization.decoders import decode_asset_like
- return cls(decode_asset_like(data["asset_condition"]))
+ return cls(
+ decode_asset_like(data["asset_condition"]),
+ batch_asset_events=data.get("batch_asset_events", True),
+ )
@property
def summary(self) -> str:
return "Asset"
def serialize(self) -> dict[str, Any]:
- return {"asset_condition": encode_asset_like(self.asset_condition)}
+ return {
+ "asset_condition": encode_asset_like(self.asset_condition),
+ "batch_asset_events": self.batch_asset_events,
+ }
+
+ def get_batching_requirement(self) -> str | None:
+ """Return why this timetable cannot run without event batching, or ``None``."""
+ pending = [self.asset_condition]
+ while pending:
+ condition = pending.pop()
+ if isinstance(condition, SerializedAssetAll) and len(condition.objects) > 1:
Review Comment:
Can this be a property on the asset condition classes instead (AssetAll returns True when it has more than one object, boolean conditions combine their children), like `PartitionMapper.is_rollup`? Then new condition types can declare this themselves.
--
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: [PR] Make asset event batching opt-in [airflow]
Posted by "uranusjr (via GitHub)" <gi...@apache.org>.
uranusjr commented on code in PR #68517:
URL: https://github.com/apache/airflow/pull/68517#discussion_r4204091565
##########
airflow-core/docs/authoring-and-scheduling/asset-scheduling.rst:
##########
@@ -75,18 +75,20 @@ You can find a listing of the relationships between assets and Dags in the :ref:
Multiple assets
-----------------
-Because the ``schedule`` parameter is a list, Dags can require multiple assets. Airflow schedules a Dag after **all** assets
-the Dag consumes have been updated at least once since the last time the Dag ran:
+Dags can require multiple assets by combining them with ``&``; such Dags batch events by default.
+Airflow schedules a Dag after **all** assets the Dag consumes have been updated at least
+once since the last time the Dag ran:
.. code-block:: python
+ from airflow.sdk import AssetTriggeredTimetable
+
with DAG(
dag_id="multiple_assets_example",
- schedule=[
- example_asset_1,
- example_asset_2,
- example_asset_3,
- ],
+ schedule=AssetTriggeredTimetable(
+ assets=example_asset_1 & example_asset_2 & example_asset_3,
+ batch_asset_events=True,
+ ),
Review Comment:
`&` conditions already default to batching, these examples shouldn’t need to be rewritten?
--
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