You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@airflow.apache.org by "hussein-awala (via GitHub)" <gi...@apache.org> on 2024/02/09 23:23:39 UTC
[PR] Support multiple XCom output in the BaseOperator [airflow]
hussein-awala opened a new pull request, #37297:
URL: https://github.com/apache/airflow/pull/37297
closes: #37284
This PR adds a new argument `multiple_outputs ` to BaseOperator to push multiple XCom when it set to True instead of pushing the result to `return_value` XCom.
--
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] Support multiple XCom output in the BaseOperator [airflow]
Posted by "potiuk (via GitHub)" <gi...@apache.org>.
potiuk commented on code in PR #37297:
URL: https://github.com/apache/airflow/pull/37297#discussion_r1485043421
##########
airflow/models/taskinstance.py:
##########
@@ -450,7 +450,16 @@ def _execute_callable(context, **execute_callable_kwargs):
else:
xcom_value = None
if xcom_value is not None: # If the task returns a result, push an XCom containing it.
- task_instance.xcom_push(key=XCOM_RETURN_KEY, value=xcom_value, session=session)
+ if task_to_execute.multiple_outputs:
+ if not isinstance(xcom_value, Mapping):
+ raise AirflowException(
+ "The returned value from the task is not a Mapping. The `multiple_outputs`"
+ "parameter is set to True, so the result has to be a Mapping."
+ )
+ for key, value in xcom_value.items():
Review Comment:
We do not limit the size in single output, so I think we should not do it in multiple output case either.
--
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] Support multiple XCom output in the BaseOperator [airflow]
Posted by "jannisko (via GitHub)" <gi...@apache.org>.
jannisko commented on PR #37297:
URL: https://github.com/apache/airflow/pull/37297#issuecomment-1936935927
Thanks for jumping on this request! @hussein-awala
I was kind of hoping to contribute myself but I guess I was too slow 😄
--
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] Support multiple XCom output in the BaseOperator [airflow]
Posted by "rawwar (via GitHub)" <gi...@apache.org>.
rawwar commented on code in PR #37297:
URL: https://github.com/apache/airflow/pull/37297#discussion_r1484897249
##########
airflow/models/taskinstance.py:
##########
@@ -450,7 +450,16 @@ def _execute_callable(context, **execute_callable_kwargs):
else:
xcom_value = None
if xcom_value is not None: # If the task returns a result, push an XCom containing it.
- task_instance.xcom_push(key=XCOM_RETURN_KEY, value=xcom_value, session=session)
+ if task_to_execute.multiple_outputs:
+ if not isinstance(xcom_value, Mapping):
+ raise AirflowException(
+ "The returned value from the task is not a Mapping. The `multiple_outputs`"
+ "parameter is set to True, so the result has to be a Mapping."
+ )
+ for key, value in xcom_value.items():
Review Comment:
Wondering if we should limit number of outputs here?
Not really sure if this will cause issues. But, what if one accidentally returned so much data or maybe just a very large list of values. Would that cause performance issues?
--
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] Support multiple XCom output in the BaseOperator [airflow]
Posted by "rawwar (via GitHub)" <gi...@apache.org>.
rawwar commented on code in PR #37297:
URL: https://github.com/apache/airflow/pull/37297#discussion_r1484897715
##########
airflow/models/taskinstance.py:
##########
@@ -450,7 +450,16 @@ def _execute_callable(context, **execute_callable_kwargs):
else:
xcom_value = None
if xcom_value is not None: # If the task returns a result, push an XCom containing it.
- task_instance.xcom_push(key=XCOM_RETURN_KEY, value=xcom_value, session=session)
+ if task_to_execute.multiple_outputs:
+ if not isinstance(xcom_value, Mapping):
+ raise AirflowException(
+ "The returned value from the task is not a Mapping. The `multiple_outputs`"
+ "parameter is set to True, so the result has to be a Mapping."
+ )
+ for key, value in xcom_value.items():
Review Comment:
An alternative I can think of is using `return_value_count`(or probably some good variable name for this) instead of just `multiple_outputs`. But, just a suggestion.
--
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] Support multiple XCom output in the BaseOperator [airflow]
Posted by "rawwar (via GitHub)" <gi...@apache.org>.
rawwar commented on code in PR #37297:
URL: https://github.com/apache/airflow/pull/37297#discussion_r1485049460
##########
airflow/models/taskinstance.py:
##########
@@ -450,7 +450,16 @@ def _execute_callable(context, **execute_callable_kwargs):
else:
xcom_value = None
if xcom_value is not None: # If the task returns a result, push an XCom containing it.
- task_instance.xcom_push(key=XCOM_RETURN_KEY, value=xcom_value, session=session)
+ if task_to_execute.multiple_outputs:
+ if not isinstance(xcom_value, Mapping):
+ raise AirflowException(
+ "The returned value from the task is not a Mapping. The `multiple_outputs`"
+ "parameter is set to True, so the result has to be a Mapping."
+ )
+ for key, value in xcom_value.items():
Review Comment:
I don't see an option to resolve the conversation. please mark this as resolved.
--
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] Support multiple XCom output in the BaseOperator [airflow]
Posted by "jannisko (via GitHub)" <gi...@apache.org>.
jannisko commented on PR #37297:
URL: https://github.com/apache/airflow/pull/37297#issuecomment-1936972290
The implementation for the taskflow decorator seems to return the xcoms separately (with different keys) **and** in its original form with the default key:
<img width="768" alt="image" src="https://github.com/apache/airflow/assets/40455076/6cd4f3c1-2858-49ca-a3ab-a85f2d8862d1">
I'm not sure if it makes sense to recreate this behavior.
--
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] Support multiple XCom output in the BaseOperator [airflow]
Posted by "potiuk (via GitHub)" <gi...@apache.org>.
potiuk commented on PR #37297:
URL: https://github.com/apache/airflow/pull/37297#issuecomment-1937064243
> I didn't actually notice that there was type-hint based inferral until now 😄
> I feel like this feature is too niche to warrant introducing a breaking change to all operators though.
Yep. Author of the classic operator should make the decision on their own if they want to return multiple_outputs or not.
--
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] Support multiple XCom output in the BaseOperator [airflow]
Posted by "rawwar (via GitHub)" <gi...@apache.org>.
rawwar commented on code in PR #37297:
URL: https://github.com/apache/airflow/pull/37297#discussion_r1484897715
##########
airflow/models/taskinstance.py:
##########
@@ -450,7 +450,16 @@ def _execute_callable(context, **execute_callable_kwargs):
else:
xcom_value = None
if xcom_value is not None: # If the task returns a result, push an XCom containing it.
- task_instance.xcom_push(key=XCOM_RETURN_KEY, value=xcom_value, session=session)
+ if task_to_execute.multiple_outputs:
+ if not isinstance(xcom_value, Mapping):
+ raise AirflowException(
+ "The returned value from the task is not a Mapping. The `multiple_outputs`"
+ "parameter is set to True, so the result has to be a Mapping."
+ )
+ for key, value in xcom_value.items():
Review Comment:
An alternative I can think of is using `return_value_count` instead of just `multiple_outputs`. But, just a suggestion.
--
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] Support multiple XCom output in the BaseOperator [airflow]
Posted by "Taragolis (via GitHub)" <gi...@apache.org>.
Taragolis commented on code in PR #37297:
URL: https://github.com/apache/airflow/pull/37297#discussion_r1485684905
##########
airflow/providers/sftp/decorators/sensors/sftp.py:
##########
@@ -52,7 +52,6 @@ def __init__(
task_id: str,
**kwargs,
) -> None:
- kwargs.pop("multiple_outputs")
Review Comment:
Is this change break this decorator in case of Airflow < 2.9?
Maybe we should make it under the condition?
--
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] Support multiple XCom output in the BaseOperator [airflow]
Posted by "jannisko (via GitHub)" <gi...@apache.org>.
jannisko commented on code in PR #37297:
URL: https://github.com/apache/airflow/pull/37297#discussion_r1485018215
##########
airflow/models/taskinstance.py:
##########
@@ -450,7 +450,16 @@ def _execute_callable(context, **execute_callable_kwargs):
else:
xcom_value = None
if xcom_value is not None: # If the task returns a result, push an XCom containing it.
- task_instance.xcom_push(key=XCOM_RETURN_KEY, value=xcom_value, session=session)
+ if task_to_execute.multiple_outputs:
+ if not isinstance(xcom_value, Mapping):
+ raise AirflowException(
+ "The returned value from the task is not a Mapping. The `multiple_outputs`"
+ "parameter is set to True, so the result has to be a Mapping."
+ )
+ for key, value in xcom_value.items():
Review Comment:
I would argue since it's an explicit parameter, the dag author should be responsible for making sure they aren't overloading the xcom. I believe that Airflow already doesn't limit xcom sizes, but instead it's just a warning in the documentation. It could be the same for the split values.
There is also no limit in the existing implementation for the decorator tasks: https://github.com/apache/airflow/blob/8fac799a250a222bf85d484947a9b3f2065d5ba4/airflow/decorators/base.py#L244
--
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] Support multiple XCom output in the BaseOperator [airflow]
Posted by "hussein-awala (via GitHub)" <gi...@apache.org>.
hussein-awala merged PR #37297:
URL: https://github.com/apache/airflow/pull/37297
--
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] Support multiple XCom output in the BaseOperator [airflow]
Posted by "jannisko (via GitHub)" <gi...@apache.org>.
jannisko commented on PR #37297:
URL: https://github.com/apache/airflow/pull/37297#issuecomment-2252886312
@joycechan-KSO yes this feature was available for the [TaskFlow](https://airflow.apache.org/docs/apache-airflow/stable/tutorial/taskflow.html) style operators already, but missing for the classic operators inheriting from `BaseOperator` (see [original issue](https://github.com/apache/airflow/issues/37284)) 🙂
--
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] Support multiple XCom output in the BaseOperator [airflow]
Posted by "jannisko (via GitHub)" <gi...@apache.org>.
jannisko commented on code in PR #37297:
URL: https://github.com/apache/airflow/pull/37297#discussion_r1485018215
##########
airflow/models/taskinstance.py:
##########
@@ -450,7 +450,16 @@ def _execute_callable(context, **execute_callable_kwargs):
else:
xcom_value = None
if xcom_value is not None: # If the task returns a result, push an XCom containing it.
- task_instance.xcom_push(key=XCOM_RETURN_KEY, value=xcom_value, session=session)
+ if task_to_execute.multiple_outputs:
+ if not isinstance(xcom_value, Mapping):
+ raise AirflowException(
+ "The returned value from the task is not a Mapping. The `multiple_outputs`"
+ "parameter is set to True, so the result has to be a Mapping."
+ )
+ for key, value in xcom_value.items():
Review Comment:
I would argue since it's an explicit parameter, the dag author should be responsible for making sure they aren't overloading the xcom. I believe that Airflow already doesn't limit xcom sizes, but instead it's just a warning in the documentation. It could be the same for the split values.
There is also no limit in the existing implementation for the decorator tasks: https://github.com/apache/airflow/blob/8fac799a250a222bf85d484947a9b3f2065d5ba4/airflow/decorators/base.py#L244-L277
--
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] Support multiple XCom output in the BaseOperator [airflow]
Posted by "jannisko (via GitHub)" <gi...@apache.org>.
jannisko commented on PR #37297:
URL: https://github.com/apache/airflow/pull/37297#issuecomment-1937048094
> Yes, we can do that. I wonder if we need to duplicate all the behaviors, for example, inferring `multiple_outputs` (#20800) when the execute method has a mapping as the returned value type.
I didn't actually notice that there was type-hint based inferral until now 😄
I feel like this feature is too niche to warrant introducing a breaking change to all operators though.
--
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] Support multiple XCom output in the BaseOperator [airflow]
Posted by "hussein-awala (via GitHub)" <gi...@apache.org>.
hussein-awala commented on PR #37297:
URL: https://github.com/apache/airflow/pull/37297#issuecomment-1937021429
> I'm not sure if it makes sense to recreate this behavior.
Yes, we can do that. I wonder if we need to duplicate all the behaviors, for example, inferring `multiple_outputs` (https://github.com/apache/airflow/pull/20800) when the execute method has a mapping as the returned value type.
--
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] Support multiple XCom output in the BaseOperator [airflow]
Posted by "joycechan-KSO (via GitHub)" <gi...@apache.org>.
joycechan-KSO commented on PR #37297:
URL: https://github.com/apache/airflow/pull/37297#issuecomment-2250333988
Hi, I think this feature been added by airflow 2.0 + version
https://airflow.apache.org/docs/apache-airflow/2.0.0/concepts.html
is this feature dropped before?
--
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] Support multiple XCom output in the BaseOperator [airflow]
Posted by "hussein-awala (via GitHub)" <gi...@apache.org>.
hussein-awala commented on code in PR #37297:
URL: https://github.com/apache/airflow/pull/37297#discussion_r1485695577
##########
airflow/providers/sftp/decorators/sensors/sftp.py:
##########
@@ -52,7 +52,6 @@ def __init__(
task_id: str,
**kwargs,
) -> None:
- kwargs.pop("multiple_outputs")
Review Comment:
You're right, I will do that in a separate PR once this one is merged, for now I will just revert 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