You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@airflow.apache.org by "sjyangkevin (via GitHub)" <gi...@apache.org> on 2025/05/05 01:47:00 UTC
[PR] PR to Issue 39017: Allow DAGs to trigger on Datasets with wildcard/regex [airflow]
sjyangkevin opened a new pull request, #50182:
URL: https://github.com/apache/airflow/pull/50182
The PR is to approach the use cases mentioned in the issue #39017. The idea is communicated in the issue.
The design decision and implementation plan to approach the use case.
https://github.com/apache/airflow/issues/39017#issuecomment-2849005521
Opening this draft PR to get early feedback.
<!--
Licensed to the Apache Software Foundation (ASF) under one
or more contributor license agreements. See the NOTICE file
distributed with this work for additional information
regarding copyright ownership. The ASF licenses this file
to you under the Apache License, Version 2.0 (the
"License"); you may not use this file except in compliance
with the License. You may obtain a copy of the License at
http://www.apache.org/licenses/LICENSE-2.0
Unless required by applicable law or agreed to in writing,
software distributed under the License is distributed on an
"AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
KIND, either express or implied. See the License for the
specific language governing permissions and limitations
under the License.
-->
<!--
Thank you for contributing! 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.
Feel free to ping committers for the review!
In case of an existing issue, reference it using one of the following:
closes: #ISSUE
related: #ISSUE
How to write a good git commit message:
http://chris.beams.io/posts/git-commit/
-->
<!-- Please keep an empty line above the dashes. -->
---
**^ Add meaningful description above**
Read the **[Pull Request Guidelines](https://github.com/apache/airflow/blob/main/contributing-docs/05_pull_requests.rst#pull-request-guidelines)** for more information.
In case of fundamental code changes, an Airflow Improvement Proposal ([AIP](https://cwiki.apache.org/confluence/display/AIRFLOW/Airflow+Improvement+Proposals)) is needed.
In case of a new dependency, check compliance with the [ASF 3rd Party License Policy](https://www.apache.org/legal/resolved.html#category-x).
In case of backwards incompatible changes please leave a note in a newsfragment file, named `{pr_number}.significant.rst` or `{issue_number}.significant.rst`, in [airflow-core/newsfragments](https://github.com/apache/airflow/tree/main/airflow-core/newsfragments).
--
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] Propagate AssetEvent through context when Consumer DAG triggering on AssetAlias [airflow]
Posted by "sjyangkevin (via GitHub)" <gi...@apache.org>.
sjyangkevin commented on PR #50182:
URL: https://github.com/apache/airflow/pull/50182#issuecomment-2926666261
Want to share some small updates. I've merged the two queries into one and run all the unit tests in jobs/, create an actual DAG to test the behavior, and profile the query runtime. I observed the following behaviours for the DAG (attached at the end of the comment)
1. I use `asset_1 | asset_2 | asset_alias` to schedule a consumer DAG. I have a producer DAG emit all the events for both regular asset events, and asset events attached to the alias at runtime. This results in two consumer DAG runs, one for regular asset, and another one for asset alias (include all asset events attached).
2. Both asset and asset alias can carry the asset event to the consumer DAG.
Since the query use multiple `.any`, I am a little wondering about the performance. So, I have the producer DAG first create 100 asset events and attached to the alias.
### Two queries (the first implementation)
Since it results in two DAG Runs, the first one captures 2 asset events of the regular `Asset`, the runtime is 0.0065 seconds. Then, the second one captures 100 asset events attached to the alias, the runtime is 0.0063 seconds.
### One query (merged, current implementation)
The runtime is 0.0050 seconds and 0.0060 seconds.
### Producer DAG
```python
from airflow.sdk import Asset, AssetAlias, DAG
from airflow.providers.standard.operators.python import PythonOperator
from airflow.providers.standard.operators.empty import EmptyOperator
asset_alias = AssetAlias("asset_alias_massive_events")
asset_1 = Asset("my_asset_1")
asset_2 = Asset("my_asset_2")
with DAG(
dag_id="profile_asset_query_producer",
):
def _generate_regular_asset_event():
print("generate 100 asset events for regular assets")
def _attach_asset_event_to_alias(**context):
outlet_events = context["outlet_events"]
for i in range(1, 101): # I also tried 1001, generate 1000 DAGs
outlet_events[asset_alias].add(
Asset(f"my_asset_alias_{i}"),
extra={"filename": f"my_asset_alias_{i}"}
)
start = EmptyOperator(task_id="start")
generate_regular_asset_event = PythonOperator(
task_id="generate_regular_asset_event",
python_callable=_generate_regular_asset_event,
outlets=[asset_1, asset_2]
)
attach_asset_event_to_alias = PythonOperator(
task_id="attach_asset_event_to_alias",
python_callable=_attach_asset_event_to_alias,
outlets=[asset_alias]
)
end = EmptyOperator(task_id="end")
start >> [generate_regular_asset_event, attach_asset_event_to_alias] >> end
```
### Consumer DAG
```python
from airflow.sdk import Asset, AssetAlias, DAG
from airflow.providers.standard.operators.python import PythonOperator
from airflow.providers.standard.operators.empty import EmptyOperator
asset_alias = AssetAlias("asset_alias_massive_events")
asset_1 = Asset("my_asset_1")
asset_2 = Asset("my_asset_2")
with DAG(
dag_id="profile_asset_query_consumer",
schedule=(asset_1 | asset_2 | asset_alias)
):
def _log_asset_events(**context):
print("context:", context)
log_asset_events = PythonOperator(
task_id="log_asset_events",
python_callable=_log_asset_events,
inlets=[asset_1, asset_2, asset_alias]
)
```
--
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] Propagate AssetEvent through context when Consumer DAG triggering on AssetAlias [airflow]
Posted by "Lee-W (via GitHub)" <gi...@apache.org>.
Lee-W commented on code in PR #50182:
URL: https://github.com/apache/airflow/pull/50182#discussion_r2131368876
##########
airflow-core/tests/unit/jobs/test_scheduler_job.py:
##########
@@ -3995,6 +4001,82 @@ def dict_from_obj(obj):
assert created_run.creating_job_id == scheduler_job.id
+ @pytest.mark.need_serialized_dag
+ def test_create_dag_runs_asset_alias_with_asset_event_attached(self, session, dag_maker):
+ """
+ Test DAG Run trigger on AssetAlias includes the corresponding AssetEvent in `consumed_asset_events`.
Review Comment:
```suggestion
Test Dag Run trigger on AssetAlias includes the corresponding AssetEvent in `consumed_asset_events`.
```
nit
##########
airflow-core/tests/unit/jobs/test_scheduler_job.py:
##########
@@ -3995,6 +4001,82 @@ def dict_from_obj(obj):
assert created_run.creating_job_id == scheduler_job.id
+ @pytest.mark.need_serialized_dag
+ def test_create_dag_runs_asset_alias_with_asset_event_attached(self, session, dag_maker):
+ """
+ Test DAG Run trigger on AssetAlias includes the corresponding AssetEvent in `consumed_asset_events`.
+ """
+
+ # Simulate an Asset created at runtime, and it is not an active asset
+ asset1 = Asset(uri="test://asset1", name="test_asset", group="test_group")
+ # Create an AssetAlias, and the Asset will be attached to this AssetAlias
+ asset_alias = AssetAlias(name="test_asset_alias_with_asset_event", group="test_group")
+
+ # Add it to the DB so the event can be created from this Asset
+ asm = AssetModel(name=asset1.name, uri=asset1.uri, group=asset1.group)
+ session.add(asm)
+
+ asam = AssetAliasModel(name=asset_alias.name, group=asset_alias.group)
+
+ # Simulate a Producer DAG attach an asset event at runtime to an AssetAlias
Review Comment:
```suggestion
# Simulate a Producer dag attach an asset event at runtime to an AssetAlias
```
--
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] Propagate AssetEvent through context when Consumer DAG triggering on AssetAlias [airflow]
Posted by "sjyangkevin (via GitHub)" <gi...@apache.org>.
sjyangkevin commented on PR #50182:
URL: https://github.com/apache/airflow/pull/50182#issuecomment-2947913195
> Overall, it looks good to me! Thanks!
Thank 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] Allow DAGs to trigger on Datasets with wildcard/regex [airflow]
Posted by "sjyangkevin (via GitHub)" <gi...@apache.org>.
sjyangkevin commented on PR #50182:
URL: https://github.com/apache/airflow/pull/50182#issuecomment-2861250325
> > Found a unit test case `airflow-core/tests/unit/models/test_taskinstance.py`, from my understanding the extra information should be available from the `inlet_events`, since the test passed. The test is implemented in task flow api, but seems my code is in traditional syntax. Will need further investigate into it.
>
> for using traditional syntax, you can access the value through `context`
>
> https://airflow.apache.org/docs/apache-airflow/stable/release_notes.html#improved-inlet-events-outlet-events-and-triggering-asset-events
Hi Wei, thank you. I modified the DAG code to access the asset events from context based on the document you provided. However, the asset event associated with the alias is not in the `inlets`. I also tried to use TaskFlow API, but the get into the same situation. I've attached the DAG code in the end.
Also, I could try to explain it from the source code. From line 1606 to 1634
in `airflow-core/src/airflow/jobs/scheduler_job_runner.py`.
```python
# here, we only query on events join with `DagScheduleAssetReference`, so I think when the DAG is scheduled on asset
# alias, this query return empty results, and hence not populate the `consumed_asset_events`
asset_events = session.scalars(
select(AssetEvent)
.join(
DagScheduleAssetReference,
AssetEvent.asset_id == DagScheduleAssetReference.asset_id,
)
.where(
DagScheduleAssetReference.dag_id == dag.dag_id,
AssetEvent.timestamp <= triggered_date,
AssetEvent.timestamp > func.coalesce(cte.c.previous_dag_run_run_after, date.min),
)
).all()
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)
session.execute(delete(AssetDagRunQueue).where(AssetDagRunQueue.target_dag_id == dag_run.dag_id))
```
In `taskinstance.py`, the inlets seems only included the asset information if it is triggered by asset instead of asset alias.
Consumer DAG
```python
from airflow.sdk import Asset, AssetAlias, DAG
from airflow.providers.standard.operators.python import PythonOperator
file_asset_alias = AssetAlias("file_asset_event")
with DAG(
dag_id="asset_alias_consumer",
schedule=[file_asset_alias]
):
def my_function(**context):
print(context)
inlet_events = context["inlet_events"]
outlet_events = context["outlet_events"]
print("Access triggering_asset_events in the consumer task")
print("triggering_asset_events:", context["triggering_asset_events"])
print("Access inlet_events in the consumer task, from asset alias 'file_asset_event'")
print("inlet_events:", inlet_events[AssetAlias(name="file_asset_event")])
print("inlet_events:", inlet_events.for_asset_alias(name="file_asset_event"))
print("Access outlet_events in the consumer task, from asset alias 'file_asset_event'")
print("outlet_events:", outlet_events[AssetAlias(name="file_asset_event")])
print("outlet_events:", outlet_events.for_asset_alias(name="file_asset_event"))
my_task = PythonOperator(
task_id="consumer_task",
python_callable=my_function,
inlets=[file_asset_alias]
)
```
--
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] Propagate AssetEvent through context when Consumer DAG triggering on AssetAlias [airflow]
Posted by "sjyangkevin (via GitHub)" <gi...@apache.org>.
sjyangkevin commented on PR #50182:
URL: https://github.com/apache/airflow/pull/50182#issuecomment-2910934078
> > Probably we can create an API endpoint to clean up those unused assets related to an asset alias.
>
> Sounds like a good idea. I think this is something we thought of but decide to defer it till someone asks đź‘€
May I ask why this idea is deferred? I think most of the assets will be active, if those are directly assigned to an outlet or inlet. By when using AssetAlias, and Asset created at runtime, which might not be the case. Also, I am wondering if removing these assets will result in problem in scheduling if somehow a ask trying to look for that asset. Whether there is linkage to other tables that we should also clean up when removing the asset.
--
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] Allow DAGs to trigger on Datasets with wildcard/regex [airflow]
Posted by "sjyangkevin (via GitHub)" <gi...@apache.org>.
sjyangkevin commented on PR #50182:
URL: https://github.com/apache/airflow/pull/50182#issuecomment-2857061629
Found a unit test case `airflow-core/tests/unit/models/test_taskinstance.py`, from my understanding the extra information should be available from the `inlet_events`, since the test passed. The test is implemented in task flow api, but seems my code is in traditional syntax. Will need further investigate into it.
--
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] Propagate AssetEvent through context when Consumer DAG triggering on AssetAlias [airflow]
Posted by "obarisk (via GitHub)" <gi...@apache.org>.
obarisk commented on PR #50182:
URL: https://github.com/apache/airflow/pull/50182#issuecomment-3272735105
@Lee-W
On airflow 3.0.6, there's no `triggering_asset_events` for a task in a AssetAlias trigger dag.
Can we backport this to airflow 3.0+? 🙏🏼
Here's the code snippet.
```python
from pprint import pprint
from typing import Any
from airflow.sdk import Asset, AssetAlias, Context, dag, task
from airflow.sdk.execution_time.context import (
OutletEventAccessors,
TriggeringAssetEventsAccessor,
)
@dag(schedule=None, tags=["demo"])
def assetalias_producer():
@task(outlets=[AssetAlias("asset_alias_0")])
def producer(**context: Context):
outlets: list[Any] = context["outlets"]
outlet_events: OutletEventAccessors = context["outlet_events"]
run_after = context["dag_run"].run_after
for a in outlets:
pprint(a)
match a:
case AssetAlias():
if int(run_after.timestamp()) % 2 == 0:
outlet_events[a].add(Asset("local0", extra={"run_after": run_after}))
else:
outlet_events[a].add(Asset("local1", extra={"n": 1}))
outlet_events[a].add(Asset("local2", extra={"run_after": run_after}))
case Asset():
outlet_events[a].extra = {"rc": 100}
case _:
raise ValueError(f"unexpected outlet type: {type(a)}")
return 0
producer()
@dag(schedule=[AssetAlias("asset_alias_0")], tags=["demo"])
def assetalias_consumer():
@task
def consumer(**context: Context):
triggering_events: TriggeringAssetEventsAccessor = context[
"triggering_asset_events"
]
pprint(triggering_events[AssetAlias("asset_alias_0")])
consumer()
assetalias_producer()
assetalias_consumer()
```
--
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] PR to Issue 39017: Allow DAGs to trigger on Datasets with wildcard/regex [airflow]
Posted by "sjyangkevin (via GitHub)" <gi...@apache.org>.
sjyangkevin commented on PR #50182:
URL: https://github.com/apache/airflow/pull/50182#issuecomment-2849885150
The following unit tests have been run and all passed.
```
pytest --log-cli-level=DEBUG airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_assets.py
```
```
pytest --log-cli-level=DEBUG airflow-core/tests/unit/assets/
```
```
pytest --log-cli-level=DEBUG airflow-core/tests/unit/api_fastapi/execution_api/versions/head/test_assets.py
```
```
pytest --log-cli-level=DEBUG airflow-core/tests/unit/api_fastapi/execution_api/versions/head/test_asset_events.py
```
```
pytest --log-cli-level=DEBUG airflow-core/tests/unit/core/test_core.py::TestCore
```
--
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] Propagate AssetEvent through context when Consumer DAG triggering on AssetAlias [airflow]
Posted by "Lee-W (via GitHub)" <gi...@apache.org>.
Lee-W commented on PR #50182:
URL: https://github.com/apache/airflow/pull/50182#issuecomment-2910865474
> Probably we can create an API endpoint to clean up those unused assets related to an asset alias.
Sounds like a good idea. I think this is something we thought of but decide to defer it till someone asks đź‘€
--
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] Propagate AssetEvent through context when Consumer DAG triggering on AssetAlias [airflow]
Posted by "Lee-W (via GitHub)" <gi...@apache.org>.
Lee-W commented on PR #50182:
URL: https://github.com/apache/airflow/pull/50182#issuecomment-2910864550
> When an asset is created at runtime and attached to an asset alias, the asset is not set in the “outlets” or “inlets”, so the created asset will be inactive.
By "inactive," do you mean that this asset will no longer be used, rather than becoming invalid?
--
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] PR to Issue 39017: Allow DAGs to trigger on Datasets with wildcard/regex [airflow]
Posted by "boring-cyborg[bot] (via GitHub)" <gi...@apache.org>.
boring-cyborg[bot] commented on PR #50182:
URL: https://github.com/apache/airflow/pull/50182#issuecomment-2849724536
Congratulations on your first Pull Request and welcome to the Apache Airflow community! If you have any issues or are unsure about any anything please check our Contributors' Guide (https://github.com/apache/airflow/blob/main/contributing-docs/README.rst)
Here are some useful points:
- Pay attention to the quality of your code (ruff, mypy and type annotations). Our [pre-commits]( https://github.com/apache/airflow/blob/main/contributing-docs/08_static_code_checks.rst#prerequisites-for-pre-commit-hooks) will help you with that.
- In case of a new feature add useful documentation (in docstrings or in `docs/` directory). Adding a new operator? Check this short [guide](https://github.com/apache/airflow/blob/main/airflow-core/docs/howto/custom-operator.rst) Consider adding an example DAG that shows how users should use it.
- Consider using [Breeze environment](https://github.com/apache/airflow/blob/main/dev/breeze/doc/README.rst) for testing locally, it's a heavy docker but it ships with a working Airflow and a lot of integrations.
- Be patient and persistent. It might take some time to get a review or get the final approval from Committers.
- Please follow [ASF Code of Conduct](https://www.apache.org/foundation/policies/conduct) for all communication including (but not limited to) comments on Pull Requests, Mailing list and Slack.
- Be sure to read the [Airflow Coding style]( https://github.com/apache/airflow/blob/main/contributing-docs/05_pull_requests.rst#coding-style-and-best-practices).
- Always keep your Pull Requests rebased, otherwise your build might fail due to changes not related to your commits.
Apache Airflow is a community-driven project and together we are making it better 🚀.
In case of doubts contact the developers at:
Mailing List: dev@airflow.apache.org
Slack: https://s.apache.org/airflow-slack
--
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] Propagate AssetEvent through context when Consumer DAG triggering on AssetAlias [airflow]
Posted by "sjyangkevin (via GitHub)" <gi...@apache.org>.
sjyangkevin commented on PR #50182:
URL: https://github.com/apache/airflow/pull/50182#issuecomment-2910932352
> > When an asset is created at runtime and attached to an asset alias, the asset is not set in the “outlets” or “inlets”, so the created asset will be inactive.
>
> By "inactive," do you mean that this asset will no longer be used, rather than becoming invalid?
In the metadata store, there is a table for tracking active asset, if I remembered correctly. By "inactive" means the asset is not in that table. I can double-check it when I revisit the code.
--
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] Propagate AssetEvent through context when Consumer DAG triggering on AssetAlias [airflow]
Posted by "Lee-W (via GitHub)" <gi...@apache.org>.
Lee-W commented on PR #50182:
URL: https://github.com/apache/airflow/pull/50182#issuecomment-3282278100
I'm going to backport anyway đź‘€
--
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] Propagate AssetEvent through context when Consumer DAG triggering on AssetAlias [airflow]
Posted by "Lee-W (via GitHub)" <gi...@apache.org>.
Lee-W commented on PR #50182:
URL: https://github.com/apache/airflow/pull/50182#issuecomment-3276739383
I just noticed we might not have 3.0.7 as 3.1.0 beta has been released. But this will be included in 3.1.0
--
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] Propagate AssetEvent through context when Consumer DAG triggering on AssetAlias [airflow]
Posted by "Lee-W (via GitHub)" <gi...@apache.org>.
Lee-W merged PR #50182:
URL: https://github.com/apache/airflow/pull/50182
--
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] Propagate AssetEvent through context when Consumer DAG triggering on AssetAlias [airflow]
Posted by "sjyangkevin (via GitHub)" <gi...@apache.org>.
sjyangkevin commented on PR #50182:
URL: https://github.com/apache/airflow/pull/50182#issuecomment-2980059902
> I think this makes sense
Thank you very much for reviewing it and really happy to hear it is the right direction to go!
--
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] Propagate AssetEvent through context when Consumer DAG triggering on AssetAlias [airflow]
Posted by "sjyangkevin (via GitHub)" <gi...@apache.org>.
sjyangkevin commented on PR #50182:
URL: https://github.com/apache/airflow/pull/50182#issuecomment-2910936663
> In general, the idea looks good. Producing runtime assets was the reason why we implemented asset_alias back then. But making it able to consume asset event this way looks like a good idea đź‘Ť
Thanks for the feedback! Yeah, I think it make sense for attaching asset at runtime and emit asset event. I feel this is what can make the asset alias as powerful as it is expected. I feel that it solves the problem like "I want to use a set of assets to trigger my DAG, so I use asset alias. When my DAG is triggered, how do i know which asset/asset event trigger it. Should I treat all those assets in the same way. Or each should be treated differently."
--
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] Propagate AssetEvent through context when Consumer DAG triggering on AssetAlias [airflow]
Posted by "Lee-W (via GitHub)" <gi...@apache.org>.
Lee-W commented on PR #50182:
URL: https://github.com/apache/airflow/pull/50182#issuecomment-2910865839
> Probably we can create an API endpoint to clean up those unused assets related to an asset alias.
Sounds like a good idea. I think this is something we thought of but decide to defer it till someone asks đź‘€
--
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] Propagate AssetEvent through context when Consumer DAG triggering on AssetAlias [airflow]
Posted by "sjyangkevin (via GitHub)" <gi...@apache.org>.
sjyangkevin commented on PR #50182:
URL: https://github.com/apache/airflow/pull/50182#issuecomment-2946456511
Hi @Lee-W and @uranusjr , don’t want to rush you, but I wanted to kindly follow up on any feedback if you could have some time to review it. I am planning to add more explanation about the change in query. Feel free to let me know if there is anything I can put in to help the review. Very appreciate your time
--
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] Propagate AssetEvent through context when Consumer DAG triggering on AssetAlias [airflow]
Posted by "Lee-W (via GitHub)" <gi...@apache.org>.
Lee-W commented on PR #50182:
URL: https://github.com/apache/airflow/pull/50182#issuecomment-2968926455
@uranusjr I'm thinking of merging this, but I'd like to know your thoughts on this one. Thanks!
--
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] Allow DAGs to trigger on Datasets with wildcard/regex [airflow]
Posted by "Lee-W (via GitHub)" <gi...@apache.org>.
Lee-W commented on PR #50182:
URL: https://github.com/apache/airflow/pull/50182#issuecomment-2861121054
> Found a unit test case `airflow-core/tests/unit/models/test_taskinstance.py`, from my understanding the extra information should be available from the `inlet_events`, since the test passed. The test is implemented in task flow api, but seems my code is in traditional syntax. Will need further investigate into it.
for using traditional syntax, you can access the value through `context`
--
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] Allow DAGs to trigger on Datasets with wildcard/regex [airflow]
Posted by "sjyangkevin (via GitHub)" <gi...@apache.org>.
sjyangkevin commented on PR #50182:
URL: https://github.com/apache/airflow/pull/50182#issuecomment-2853247718
> I'm a bit confused by why do we need to change the scheduler here. Could you help me understand it a bit? Thanks 🙂
>
> We use something like
>
> https://github.com/apache/airflow/blob/464da35203651308c7caa3e64a5571b32deb4a7b/airflow-core/src/airflow/models/asset.py#L75
>
> to expand an alias to assets and let them works like other assets
Thanks for the follow up. The main reason in short, is to let the consumer task able to access the corresponding asset event attached to the asset alias on which the consumer DAG Run is triggered. I also left some thinking about why it is changed in scheduler: https://github.com/apache/airflow/issues/39017#issuecomment-2853230727.
--
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] Propagate AssetEvent through context when Consumer DAG triggering on AssetAlias [airflow]
Posted by "Lee-W (via GitHub)" <gi...@apache.org>.
Lee-W commented on PR #50182:
URL: https://github.com/apache/airflow/pull/50182#issuecomment-2910974359
> In the metadata store, there is a table for tracking active asset, if I remembered correctly. By "inactive" means the asset is not in that table. I can double-check it when I revisit the code.
Means an asset is referenced or does not conflict with another asset. In Airflow 3, the design changes a bit. there's an `AssetActive` table to track it.
> May I ask why this idea is deferred? I think most of the assets will be active, if those are directly assigned to an outlet or inlet. By when using AssetAlias, and Asset created at runtime, which might not be the case. Also, I am wondering if removing these assets will result in problem in scheduling if somehow a ask trying to look for that asset. Whether there is linkage to other tables that we should also clean up when removing the asset.
not solid reason I think. it's just not yet asked 🤔
> Thanks for the feedback! Yeah, I think it m...
yep, it's denitily something need more discussion 🤔
--
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] Propagate AssetEvent through context when Consumer DAG triggering on AssetAlias [airflow]
Posted by "sjyangkevin (via GitHub)" <gi...@apache.org>.
sjyangkevin commented on PR #50182:
URL: https://github.com/apache/airflow/pull/50182#issuecomment-2909594188
> I think this makes sense. A few questions on the implementation:
>
> 1. I don’t think we need to explicitly join the association table. I believe SQLAlchemy has some tools to join the two models directly together ([such as `.any()`](https://stackoverflow.com/questions/6474989)). Can you look into this?
> 2. Is it possible to merge the two events select queries into one?
> 3. If the second point is not possible, is calling `extend()` OK if the two queries contain duplicates? I don’t recall how SQLAlchemy works well enough around this. It may emit a constraint conflict.
Very appreciate the feedback. Your point is totally valid.
1. yes, I definitely will look into this. I also feel that the join is redundant and there may be a better way to get the asset event information.
for 2.) and 3.), I have run a few testing with mixed Asset and AssetAlias, but maybe if one event is belong to both an Asset and an Alias. Will investigate more into it.
--
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] Allow DAGs to trigger on Datasets with wildcard/regex [airflow]
Posted by "sjyangkevin (via GitHub)" <gi...@apache.org>.
sjyangkevin commented on PR #50182:
URL: https://github.com/apache/airflow/pull/50182#issuecomment-2869167468
> Found a unit test case `airflow-core/tests/unit/models/test_taskinstance.py`, from my understanding the extra information should be available from the `inlet_events`, since the test passed. The test is implemented in task flow api, but seems my code is in traditional syntax. Will need further investigate into it.
After having a deeper look into the test cases. The AssetAlias is in the `inlet_events`, but if there is an AssetEvent attached to the AssetAlias in the producer DAG, the AssetEvent is not accessible.
--
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