You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@airflow.apache.org by GitBox <gi...@apache.org> on 2022/05/30 11:06:37 UTC
[GitHub] [airflow] uranusjr opened a new issue, #24021: Functional helpers on XComArg and task expansion against 'zipped' inputs
uranusjr opened a new issue, #24021:
URL: https://github.com/apache/airflow/issues/24021
### Body
A need has been established to expand a task from aggregate inputs from different upstream tasks in a “zipped” fashion, rather than cartesian product as implemented by `expand()`. This is currently doable for taskflow tasks, by injecting a task to perform the zip operation, and unpack the zipped values manually:
```python
@task
def up_1():
return ["a", "b"]
@task
def up_2():
return ["x", "y"]
@task
def aggregate(v1, v2): # The intermediate job to zip.
return list(zip(v1, v2))
@task
def output(v):
print(v)
aggregated = aggergate.expand(v1=up_1(), v2=up_2())
output.expand(v=aggregated)
# Creates two mapped tasks printing
# ["a", "x"]
# ["b", "y"]
```
However, the same cannot be done to a classic operator class, since Python does not offer a way to unpack the zipped item into keyword arguments.
Furthermore, the intermediate aggregation task (that performs zip) can introduce significant overhead, since it needs to create a new process, and load the entire lists from upstreams, which is one of the things task mapping aim to avoid if possible.
----
## Implement `expand_kwargs` on classic operators
To make zip unpacking possible for classic operators, we will add a new method `expand_kwargs` that works similar to `expand`, but takes one single list of dicts instead. The task expands against the list, and unpack each dict into the operator’s keyword arguments. For example:
```python
@task
def to_dict(v):
return {"namespace": v[0], "image": v[1]}
kwargs = to_dict(v=aggregated)
KubernetesPodOperator.partial(task_id="kube").expand_kwargs(kwargs)
# Produces two mapped tasks:
# KubernetesPodOperator(namespace="a", image="x")
# KubernetesPodOperator(namespace="b", image="y")
```
-----
## Implement `map()` and on `XComArg`
To reduce overhead to convert upstream values into a zipped _dict_, we will implement additional methods on `XComArg`. This will be done by adding a “modifier” callable on `XComArg`, that is called before the downstream consumer is executed, when the `XComArg` is being resolved. This means we can rewrite the above to
```python
def to_dict(v):
return {"namespace": v[0], "image": v[1]}
kwargs = aggregated.map(to_dict)
KubernetesPodOperator.partial(task_id="kube").expand_kwargs(kwargs)
# Same result as above!
```
Note that `to_dict` in this version is _not a task_. This essentially “merges” the previous `to_dict` task into the `KubernetesPodOperator` task and eliminates the processes needed for the trivial operation.
It is also conceivable to implement other functional helpers, such as `flat_map()`, to `XComArg` in the future.
### Committer
- [X] I acknowledge that I am a maintainer/committer of the Apache Airflow project.
--
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
[GitHub] [airflow] uranusjr commented on issue #24021: Functional helpers on XComArg and task expansion against 'zipped' inputs
Posted by GitBox <gi...@apache.org>.
uranusjr commented on issue #24021:
URL: https://github.com/apache/airflow/issues/24021#issuecomment-1149671057
It will need to be implemented separately (like `expand`, taskflow and classic tasks will have different semantics), but it can work.
--
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] kaxil commented on issue #24021: Functional helpers on XComArg and task expansion against 'zipped' inputs
Posted by GitBox <gi...@apache.org>.
kaxil commented on issue #24021:
URL: https://github.com/apache/airflow/issues/24021#issuecomment-1155081935
cc @Tehada
--
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] uranusjr commented on issue #24021: Functional helpers on XComArg and task expansion against 'zipped' inputs
Posted by GitBox <gi...@apache.org>.
uranusjr commented on issue #24021:
URL: https://github.com/apache/airflow/issues/24021#issuecomment-1141516682
An alternative to `expand_kwargs` is to overload `expand` like this:
```python
# Assuming: vals_a=[1, 2], vals_b=[3, 4], vals_c=[5, 6]
FooOperator.expand(a=vals_a, b=vals_b, c=vals_c)
# FooOperator(a=1, b=3, c=5)
# FooOperator(a=2, b=4, c=6)
# Assuming: kwargs1=[{"a": 1, "b": 3}, {"a": 2, "b": 4}], kwargs2=[{"c": 5}, {"c": 6}]
FooOperator.expand(kwargs1, kwargs2)
# Same result as above!
# This is an error; can't mix positional and keyword arguments in one call.
FooOperator.expand(kwargs2, a=vals_a)
```
This makes the syntax cleaner, but I’m not quite sure if it would be more confusing to users.
--
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] uranusjr commented on issue #24021: Functional helpers on XComArg and task expansion against 'zipped' inputs
Posted by GitBox <gi...@apache.org>.
uranusjr commented on issue #24021:
URL: https://github.com/apache/airflow/issues/24021#issuecomment-1208157953
This has been implemented.
--
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] uranusjr commented on issue #24021: Functional helpers on XComArg and task expansion against 'zipped' inputs
Posted by GitBox <gi...@apache.org>.
uranusjr commented on issue #24021:
URL: https://github.com/apache/airflow/issues/24021#issuecomment-1161157141
Forgot to mention here: I also wrote some documentation additions for this, with some more useful examples: #24489. Feel free to propose any suggestions there 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
[GitHub] [airflow] uranusjr closed issue #24021: Functional helpers on XComArg and task expansion against 'zipped' inputs
Posted by GitBox <gi...@apache.org>.
uranusjr closed issue #24021: Functional helpers on XComArg and task expansion against 'zipped' inputs
URL: https://github.com/apache/airflow/issues/24021
--
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] ashb commented on issue #24021: Functional helpers on XComArg and task expansion against 'zipped' inputs
Posted by GitBox <gi...@apache.org>.
ashb commented on issue #24021:
URL: https://github.com/apache/airflow/issues/24021#issuecomment-1143284207
I def think we need one of the "native" zip interfaces for TaskFlow API too -- although it is possible to work around it, part of the goal of task mapping was to make the solution to this problem be more "native" and less hacky, so having and intermediate task is sub-optimal.
You've presented a number of options here -- can you summarize with which ones you think we should do, and in what order if more than one?
--
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] kumaranp23 commented on issue #24021: Functional helpers on XComArg and task expansion against 'zipped' inputs
Posted by GitBox <gi...@apache.org>.
kumaranp23 commented on issue #24021:
URL: https://github.com/apache/airflow/issues/24021#issuecomment-1310160259
key_pairs = file_list.map(create_pairs)
AttributeError: 'XComArg' object has no attribute 'map'
map() is not implemented on XComArg? @uranusjr
--
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] uranusjr commented on issue #24021: Functional helpers on XComArg and task expansion against 'zipped' inputs
Posted by GitBox <gi...@apache.org>.
uranusjr commented on issue #24021:
URL: https://github.com/apache/airflow/issues/24021#issuecomment-1143375651
I think we should do all of the above (except the alternative API thing, which I have deleted from the post), in the order they were mentioned.
As of native zip in taskflow, I think simply having `zip` is enough:
```python
@task
def up_1():
return ["a", "b"]
@task
def up_2():
return ["x", "y"]
@task
def output(v):
first, second = v
print(first, second)
# This should print:
# a, b
# x, y
output.expand(v=up_1().zip(up_2()))
```
This feels very close to normal Python code to me.
--
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] ashb commented on issue #24021: Functional helpers on XComArg and task expansion against 'zipped' inputs
Posted by GitBox <gi...@apache.org>.
ashb commented on issue #24021:
URL: https://github.com/apache/airflow/issues/24021#issuecomment-1148619257
Does `expand_kwargs` work on Taskflow operators 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
[GitHub] [airflow] Tehada commented on issue #24021: Functional helpers on XComArg and task expansion against 'zipped' inputs
Posted by GitBox <gi...@apache.org>.
Tehada commented on issue #24021:
URL: https://github.com/apache/airflow/issues/24021#issuecomment-1155117212
@kaxil , thanks for mention -- I already looked at proposed changes and feel like `expand_kwargs` api is quite fine. I am not sure about implementation details though -- after looking at code I understood that airflow is quite a complex, interconnected system for me to contribute some meaningful code, at least for now. I will follow closely on updates and provide my feedback, where it could be relevant.
Also I understood that my ideal usecase is slightly more complicated, than dynamic creation of params for kube pods in one task and then expanding on the list of kwargs to create variable number of pods. I would also like to have an ability to yield these kwargs one by one from params creating task to instantly create pods, while other params are still in the process of generation. But this seems to be quite far from current Airflow capabilities and more like dreams about Airflow 3.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
[GitHub] [airflow] Tehada commented on issue #24021: Functional helpers on XComArg and task expansion against 'zipped' inputs
Posted by GitBox <gi...@apache.org>.
Tehada commented on issue #24021:
URL: https://github.com/apache/airflow/issues/24021#issuecomment-1155139196
Nevermind, I guess I know, how yield-like behavior could be imitated, I just need to change this whole params generating task into task, which geenerates one instance of params for kube pod and then from aggregated task trigger all dynamic amount of tasks, with each of them generating params for one pod and add pod creation task downstream.
--
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