You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@airflow.apache.org by "dstandish (via GitHub)" <gi...@apache.org> on 2023/06/15 09:28:49 UTC
[GitHub] [airflow] dstandish opened a new pull request, #31927: Add different version of chain
dstandish opened a new pull request, #31927:
URL: https://github.com/apache/airflow/pull/31927
Personally I found this one more useful.
--
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] dstandish commented on a diff in pull request #31927: Add different version of chain
Posted by "dstandish (via GitHub)" <gi...@apache.org>.
dstandish commented on code in PR #31927:
URL: https://github.com/apache/airflow/pull/31927#discussion_r1230750927
##########
airflow/models/baseoperator.py:
##########
@@ -1838,6 +1839,38 @@ def cross_downstream(
task.set_downstream(to_tasks)
+def chain_sequential(*elements: DependencyMixin | Sequence[DependencyMixin]):
+ """
+ Helper to simplify task dependency definition.
+
+ E.g.: suppose you want precedence like so::
+
+ ╭─op2─╮ ╭─op4─╮
+ op1─┤ ├─├─op5─┤─op7
+ ╰-op3─╯ ╰-op6─╯
+
+ Then you can accomplish like so::
+
+ chain_sequential(
+ op1,
+ [op2, op3],
+ [op4, op5, op6],
+ op7
+ )
+
+ Args:
+ elements: a list of operators / lists of operators
+ """
+ prev_elem = None
+ for curr_elem in elements:
+ if isinstance(curr_elem, EdgeModifier):
+ raise ValueError("Labels are not supported by chain_sequential")
+ if prev_elem is not None:
+ for task in prev_elem:
+ task >> curr_elem
+ prev_elem = curr_elem if isinstance(curr_elem, list) else [curr_elem]
Review Comment:
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
[GitHub] [airflow] dstandish commented on a diff in pull request #31927: Add different version of chain
Posted by "dstandish (via GitHub)" <gi...@apache.org>.
dstandish commented on code in PR #31927:
URL: https://github.com/apache/airflow/pull/31927#discussion_r1231480992
##########
airflow/models/baseoperator.py:
##########
@@ -1838,6 +1839,38 @@ def cross_downstream(
task.set_downstream(to_tasks)
+def chain_sequential(*elements: DependencyMixin | Sequence[DependencyMixin]):
Review Comment:
i guess chain reduce grows on me a bit. unless `chain_linear` seemed better to you, maybe we can add this as `chain_reduce` and rename the other to `chain_zip`.
--
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 a diff in pull request #31927: Add different version of chain
Posted by "uranusjr (via GitHub)" <gi...@apache.org>.
uranusjr commented on code in PR #31927:
URL: https://github.com/apache/airflow/pull/31927#discussion_r1231759248
##########
airflow/models/baseoperator.py:
##########
@@ -1838,6 +1839,38 @@ def cross_downstream(
task.set_downstream(to_tasks)
+def chain_sequential(*elements: DependencyMixin | Sequence[DependencyMixin]):
Review Comment:
Even having `chain_zip` and `chain_linear` does not seem too weird to me 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
[GitHub] [airflow] dstandish commented on a diff in pull request #31927: Add different version of chain
Posted by "dstandish (via GitHub)" <gi...@apache.org>.
dstandish commented on code in PR #31927:
URL: https://github.com/apache/airflow/pull/31927#discussion_r1231969263
##########
airflow/models/baseoperator.py:
##########
@@ -1838,6 +1839,38 @@ def cross_downstream(
task.set_downstream(to_tasks)
+def chain_sequential(*elements: DependencyMixin | Sequence[DependencyMixin]):
Review Comment:
Yeah maybe I'll go with those too tomorrow....
--
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] ephraimbuddy commented on a diff in pull request #31927: Add different version of chain
Posted by "ephraimbuddy (via GitHub)" <gi...@apache.org>.
ephraimbuddy commented on code in PR #31927:
URL: https://github.com/apache/airflow/pull/31927#discussion_r1231078615
##########
airflow/models/baseoperator.py:
##########
@@ -1838,6 +1839,38 @@ def cross_downstream(
task.set_downstream(to_tasks)
+def chain_sequential(*elements: DependencyMixin | Sequence[DependencyMixin]):
Review Comment:
As this seemed similar to python's `zip_longest` even though it's not using None to compensate for unequal mapping(it can actually be argued), `chain_longest` looks good IMO.
--
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] dstandish commented on a diff in pull request #31927: Add different version of chain
Posted by "dstandish (via GitHub)" <gi...@apache.org>.
dstandish commented on code in PR #31927:
URL: https://github.com/apache/airflow/pull/31927#discussion_r1230752484
##########
airflow/models/baseoperator.py:
##########
@@ -1838,6 +1839,38 @@ def cross_downstream(
task.set_downstream(to_tasks)
+def chain_sequential(*elements: DependencyMixin | Sequence[DependencyMixin]):
Review Comment:
if you have better suggestion of name @uranusjr ... i'll take it
`chain_tasks`
`chain_sequential` is both right and wrong, as would be `chain_parallel`
`chain_the_other_way` 🤦
--
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 a diff in pull request #31927: Add different version of chain
Posted by "uranusjr (via GitHub)" <gi...@apache.org>.
uranusjr commented on code in PR #31927:
URL: https://github.com/apache/airflow/pull/31927#discussion_r1230746188
##########
airflow/models/baseoperator.py:
##########
@@ -1838,6 +1839,38 @@ def cross_downstream(
task.set_downstream(to_tasks)
+def chain_sequential(*elements: DependencyMixin | Sequence[DependencyMixin]):
+ """
+ Helper to simplify task dependency definition.
+
+ E.g.: suppose you want precedence like so::
+
+ ╭─op2─╮ ╭─op4─╮
+ op1─┤ ├─├─op5─┤─op7
+ ╰-op3─╯ ╰-op6─╯
+
+ Then you can accomplish like so::
+
+ chain_sequential(
+ op1,
+ [op2, op3],
+ [op4, op5, op6],
+ op7
+ )
+
+ Args:
+ elements: a list of operators / lists of operators
+ """
+ prev_elem = None
+ for curr_elem in elements:
+ if isinstance(curr_elem, EdgeModifier):
+ raise ValueError("Labels are not supported by chain_sequential")
+ if prev_elem is not None:
+ for task in prev_elem:
+ task >> curr_elem
+ prev_elem = curr_elem if isinstance(curr_elem, list) else [curr_elem]
Review Comment:
Maybe better to flip this?
```suggestion
prev_elem = [curr_elem] if isinstance(curr_elem, DependencyMixin) else curr_elem
```
This would more easily allow say tuples.
--
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] dstandish commented on a diff in pull request #31927: Add different version of chain
Posted by "dstandish (via GitHub)" <gi...@apache.org>.
dstandish commented on code in PR #31927:
URL: https://github.com/apache/airflow/pull/31927#discussion_r1231415784
##########
airflow/models/baseoperator.py:
##########
@@ -1838,6 +1839,38 @@ def cross_downstream(
task.set_downstream(to_tasks)
+def chain_sequential(*elements: DependencyMixin | Sequence[DependencyMixin]):
Review Comment:
i think the orig one should probably be called `chain_zip`
--
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] dstandish merged pull request #31927: Add different version of chain
Posted by "dstandish (via GitHub)" <gi...@apache.org>.
dstandish merged PR #31927:
URL: https://github.com/apache/airflow/pull/31927
--
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] dstandish commented on a diff in pull request #31927: Add different version of chain
Posted by "dstandish (via GitHub)" <gi...@apache.org>.
dstandish commented on code in PR #31927:
URL: https://github.com/apache/airflow/pull/31927#discussion_r1231392834
##########
airflow/models/baseoperator.py:
##########
@@ -1838,6 +1839,38 @@ def cross_downstream(
task.set_downstream(to_tasks)
+def chain_sequential(*elements: DependencyMixin | Sequence[DependencyMixin]):
Review Comment:
thanks for the suggestions.
well, here's the argument for sequential. "sequential" here means, each element provided will be arrowed to the next, in sequence, whether list or otherwise. maybe `chain_linear` would be another option, perhaps better cus it doesn't invoke the concept of sequences, or sequential vs parallel.
wdyt?
longest doesn't quite feel right because as you mention, we don't "fill out" anything to match the longest.
i do see the connection with reduce... because effectively we're reducing the elements with `>>`. maybe i would pick `chain_reduce` over `chain_reduced`. the thing that steers me away though is the fact that it invokes the `reduce` notion which makes it sound like it's doing something related to mapping.
--
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 a diff in pull request #31927: Add different version of chain
Posted by "uranusjr (via GitHub)" <gi...@apache.org>.
uranusjr commented on code in PR #31927:
URL: https://github.com/apache/airflow/pull/31927#discussion_r1230786528
##########
airflow/models/baseoperator.py:
##########
@@ -1838,6 +1839,38 @@ def cross_downstream(
task.set_downstream(to_tasks)
+def chain_sequential(*elements: DependencyMixin | Sequence[DependencyMixin]):
Review Comment:
~~`chain_correctly`~~
Maybe something to describe the fact each “group” is joined? One that comes immediately to me is `chain_reduced` since this is very much like how dynamic task mapping works when each mapped task is automatically joined (and looking back these really should have been consistent with each other). If I can design this from scratch I’d call _this_ `chain` and the other `chain_branched` but now we are stuck to coming up with the opposite of branching.
--
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] dstandish commented on a diff in pull request #31927: Add different version of chain
Posted by "dstandish (via GitHub)" <gi...@apache.org>.
dstandish commented on code in PR #31927:
URL: https://github.com/apache/airflow/pull/31927#discussion_r1232629952
##########
airflow/models/baseoperator.py:
##########
@@ -1838,6 +1839,38 @@ def cross_downstream(
task.set_downstream(to_tasks)
+def chain_sequential(*elements: DependencyMixin | Sequence[DependencyMixin]):
Review Comment:
ok -- i renamed to linear and i will add the "chain -> chain_zip" rename in a separate PR
--
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 a diff in pull request #31927: Add different version of chain
Posted by "uranusjr (via GitHub)" <gi...@apache.org>.
uranusjr commented on code in PR #31927:
URL: https://github.com/apache/airflow/pull/31927#discussion_r1233697796
##########
airflow/models/baseoperator.py:
##########
@@ -1838,6 +1839,38 @@ def cross_downstream(
task.set_downstream(to_tasks)
+def chain_linear(*elements: DependencyMixin | Sequence[DependencyMixin]):
+ """
+ Helper to simplify task dependency definition.
+
+ E.g.: suppose you want precedence like so::
+
+ ╭─op2─╮ ╭─op4─╮
+ op1─┤ ├─├─op5─┤─op7
+ ╰-op3─╯ ╰-op6─╯
+
+ Then you can accomplish like so::
+
+ chain_linear(
+ op1,
+ [op2, op3],
+ [op4, op5, op6],
+ op7
+ )
+
+ Args:
+ elements: a list of operators / lists of operators
Review Comment:
```suggestion
:param elements: a list of operators / lists of operators
```
--
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