You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@airflow.apache.org by "RNHTTR (via GitHub)" <gi...@apache.org> on 2023/08/03 21:52:10 UTC
[GitHub] [airflow] RNHTTR opened a new issue, #33088: Provide consumer DAGs with context about the dag run(s) of its producer(s)
RNHTTR opened a new issue, #33088:
URL: https://github.com/apache/airflow/issues/33088
### Description
As far as I can tell, there's currently no way for a consumer DAG to access information of the producer DAG(s) that triggered it. It'd be helpful if there was a `producers` [context variable](https://airflow.apache.org/docs/apache-airflow/stable/templates-ref.html#variables) available to task instances in a consumer that had information such as the producer(s)' dag run and task instance (i.e. the task instance in the producer DAG that wrote to the outlet) objects.
### Use case/motivation
See the [associated discussion](https://github.com/apache/airflow/discussions/33069#discussioncomment-6632003).
### Related issues
_No response_
### Are you willing to submit a PR?
- [ ] 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] Provide consumer DAGs with context about the dag run(s) of its producer(s) [airflow]
Posted by "uranusjr (via GitHub)" <gi...@apache.org>.
uranusjr commented on issue #33088:
URL: https://github.com/apache/airflow/issues/33088#issuecomment-3489670394
You can access the asset events that triggered to the downstream dag run with `triggering_asset_events`, and for each event, there are fields `source_task_id` `source_dag_id` etc. that provide information on the producing task instance object.
For the schedule part:
* You can use `source_dag_id` to get dag information through API if that’s what you want (this will be a text representation)
* If what you really want to do is to derive a _time period_ based on the upstream schedule, we are currently in the process to propose a structured way to do this with [asset partitions](https://cwiki.apache.org/confluence/display/AIRFLOW/AIP-76%2BAsset%2BPartitions).
* [You can always manually attach whatever information you want to the asset event when it is emitted.](https://airflow.apache.org/docs/apache-airflow/stable/authoring-and-scheduling/assets.html#attaching-extra-information-to-an-emitting-asset-event) Each event in the downstream should also have the `extra` attribute containing those manually set data.
I think this covers all use cases.
--
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
[GitHub] [airflow] scr-oath commented on issue #33088: Provide consumer DAGs with context about the dag run(s) of its producer(s)
Posted by "scr-oath (via GitHub)" <gi...@apache.org>.
scr-oath commented on issue #33088:
URL: https://github.com/apache/airflow/issues/33088#issuecomment-1666116748
Just thinking out loud, but if `TaskInstance` could have a copy of the schedule from its dag and a field (say `outlet_task_id` added to the Dataset class and populated when a task is triggered by a dataset, that might also be a good place to put it…
```python
class Dataset:
"""A Dataset is used for marking data dependencies between workflows."""
uri: str = attr.field(validator=[attr.validators.min_len(1), attr.validators.max_len(3000)])
extra: dict[str, Any] | None = None
outlet_task_id: Optional[str] = None
version: ClassVar[int] = 1
```
--
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
[GitHub] [airflow] scr-oath commented on issue #33088: Provide consumer DAGs with context about the dag run(s) of its producer(s)
Posted by "scr-oath (via GitHub)" <gi...@apache.org>.
scr-oath commented on issue #33088:
URL: https://github.com/apache/airflow/issues/33088#issuecomment-1666069942
Yes, I agree - and if there are multiple triggers such as `schedule=[Dataset("one"), Dataset("two")]` it would be quite generally useful to get information about each triggering event - I'm not sure exactly what "coordinates" exist for all the information about a dag or task "instance" but if a task has an outlet of a Dataset, then the "downstream" / "consumer" DAGs and their tasks should be able to learn about the information about the task instance that triggered them and its XCOM information.
--
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] Provide consumer DAGs with context about the dag run(s) of its producer(s) [airflow]
Posted by "sjyangkevin (via GitHub)" <gi...@apache.org>.
sjyangkevin commented on issue #33088:
URL: https://github.com/apache/airflow/issues/33088#issuecomment-2980304131
Not sure if it is helpful for this case. I recently have a PR, #50182 , which embed the asset event corresponding to each asset into the context variable. In this case, you can get source dag and run id, all information about the producer DAG and the asset. https://github.com/apache/airflow/pull/50182#issuecomment-2926666261
--
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] Provide consumer DAGs with context about the dag run(s) of its producer(s) [airflow]
Posted by "uranusjr (via GitHub)" <gi...@apache.org>.
uranusjr closed issue #33088: Provide consumer DAGs with context about the dag run(s) of its producer(s)
URL: https://github.com/apache/airflow/issues/33088
--
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
[GitHub] [airflow] scr-oath commented on issue #33088: Provide consumer DAGs with context about the dag run(s) of its producer(s)
Posted by "scr-oath (via GitHub)" <gi...@apache.org>.
scr-oath commented on issue #33088:
URL: https://github.com/apache/airflow/issues/33088#issuecomment-1666111504
It seems like - I could be wrong - `task_id` could be enough to get the exact xcom information.
So… if a `TaskInstance` is able to know how it was triggered - and, when scheduled is datasets/files, have corresponding list or map of dataset to the task_id that had it as an "outlet" that would be a reasonable solution to hook up the guts that could use the existing constructs to just xcom_pull from there as needed.
```
Pull XComs that optionally meet certain criteria.
:param key: A key for the XCom. If provided, only XComs with matching
keys will be returned. The default key is ``'return_value'``, also
available as constant ``XCOM_RETURN_KEY``. This key is automatically
given to XComs returned by tasks (as opposed to being pushed
manually). To remove the filter, pass *None*.
:param task_ids: Only XComs from tasks with matching ids will be
pulled. Pass *None* to remove the filter.
:param dag_id: If provided, only pulls XComs from this DAG. If *None*
(default), the DAG of the calling task is used.
:param map_indexes: If provided, only pull XComs with matching indexes.
If *None* (default), this is inferred from the task(s) being pulled
(see below for details).
:param include_prior_dates: If False, only XComs from the current
execution_date are returned. If *True*, XComs from previous dates
are returned as well.
```
--
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