From 12e48362fef39e16f6e40ce8ede80a200d373af8 Mon Sep 17 00:00:00 2001 From: pryce-turner Date: Wed, 6 Mar 2024 17:54:06 -0800 Subject: [PATCH 1/7] Nested parallelization first draft Signed-off-by: pryce-turner --- .../nested_parallelization.md | 170 ++++++++++++++++++ 1 file changed, 170 insertions(+) create mode 100644 docs/user_guide/advanced_composition/nested_parallelization.md diff --git a/docs/user_guide/advanced_composition/nested_parallelization.md b/docs/user_guide/advanced_composition/nested_parallelization.md new file mode 100644 index 0000000000..c65987bd37 --- /dev/null +++ b/docs/user_guide/advanced_composition/nested_parallelization.md @@ -0,0 +1,170 @@ +--- +jupytext: + cell_metadata_filter: all + formats: md:myst + main_language: python + notebook_metadata_filter: all + text_representation: + extension: .md + format_name: myst + format_version: 0.13 + jupytext_version: 1.16.1 +kernelspec: + display_name: Python 3 + language: python + name: python3 +--- + ++++ {"lines_to_next_cell": 0} + +(map_task)= + +# Nested Parallelization + +```{eval-rst} +.. tags:: Advanced +``` + +For exceptionally large or complicated workflows where dynamic workflows or map tasks aren't enough, it can be benficial to have multiple levels of parallelization. + +This is useful for multiple reasons: +1. Better code organization +2. Better code reuse +3. Better testing +4. Better debugging +5. Better monitoring - each subworkflow can be run independently and monitored independently +6. Better performance and scale - each subworkflow is executed as a separate workflow and thus can land on different flytepropeller workers and shards. This allows for better parallelism and scale. + +## Nested Dynamics + +This example shows how to break down a large workflow into smaller workflows and then compose them together to form a hierarchy. + +```python +""" +The structure for 6 items and a chunk size of 2, the workflow will be broken down as follows: + +multi_wf -> level1 -> level2 -> core_wf -> step1 -> step2 + -> core_wf -> step1 -> step2 + level2 -> core_wf -> step1 -> step2 + -> core_wf -> step1 -> step2 + level2 -> core_wf -> step1 -> step2 + -> core_wf -> step1 -> step2 +""" + +import typing +from flytekit import task, workflow, dynamic, LaunchPlan + + +@task +def step1(a: int) -> int: + return a + 1 + + +@task +def step2(a: int) -> int: + return a + 2 + + +@workflow +def core_wf(a: int) -> int: + return step2(a=step1(a=a)) + + +core_wf_lp = LaunchPlan.get_or_create(core_wf) + + +@dynamic +def level2(l: typing.List[int]) -> typing.List[int]: + return [core_wf_lp(a=a) for a in l] + + +@task +def reduce(l: typing.List[typing.List[int]]) -> typing.List[int]: + f = [] + for i in l: + f.extend(i) + return f + + +@dynamic +def level1(l: typing.List[int], chunk: int) -> typing.List[int]: + v = [] + for i in range(0, len(l), chunk): + v.append(level2(l=l[i:i + chunk])) + return reduce(l=v) + + +@workflow +def multi_wf(l: typing.List[int], chunk: int) -> typing.List[int]: + return level1(l=l, chunk=chunk) +``` + ++++ {"lines_to_next_cell": 0} + +This shows a top-level workflow which uses 2 levels of dynamic workflows to process a list through some simple addition tasks and then flatten it again. Here is a visual representation of the execution in a Flyte console: + +:::{figure} https://github.com/flyteorg/static-resources/blob/main/flytesnacks/user_guide/flyte_nested_parallelization.png?raw=true +:alt: Nested Parallelization UI View +:class: with-shadow +::: + +## Mixed Parallelism + +This example is similar to the above but instead of using a dynamic to loop through a 2 task workflow in series, we're using that workflow to call a map task which processes both inputs in parallel. + +```python +""" +The structure for 6 items and a chunk size of 2, the workflow will be broken down as follows: + +multi_wf -> level1 -> level2 -> mappable + -> mappable + level2 -> mappable + -> mappable + level2 -> mappable + -> mappable +""" +import typing +from flytekit import task, workflow, dynamic, map_task + + +@task +def mappable(a: int) -> int: + return a + 2 + + +@workflow +def level2(l: typing.List[int]) -> typing.List[int]: + return map_task(mappable)(a=l) + + +@task +def reduce(l: typing.List[typing.List[int]]) -> typing.List[int]: + f = [] + for i in l: + f.extend(i) + return f + + +@dynamic +def level1(l: typing.List[int], chunk: int) -> typing.List[int]: + v = [] + for i in range(0, len(l), chunk): + v.append(level2(l=l[i : i + chunk])) + return reduce(l=v) + + +@workflow +def multi_wf(l: typing.List[int], chunk: int) -> typing.List[int]: + return level1(l=l, chunk=chunk) + +``` ++++ {"lines_to_next_cell": 0} + +## Concurrency vs Max Parallelism + +Should we document this distinction or is it going away soon? + +## Design Considerations + +You can nest even further if needed, or incorporate map tasks if your inputs are all the same type. The design of your workflow should of course be informed by the actual data you're processing. For example if you have a big library of music that you'd like to extract the lyrics for, the first level can loop through all the albums and the second level can go through each song. + From f7a8434ff18bf7ee59eec623cd5067d654ca6ae7 Mon Sep 17 00:00:00 2001 From: pryce-turner Date: Fri, 8 Mar 2024 10:34:59 -0800 Subject: [PATCH 2/7] Added screenshots and considerations section Signed-off-by: pryce-turner --- .../nested_parallelization.md | 30 +++++++++++++++---- 1 file changed, 25 insertions(+), 5 deletions(-) diff --git a/docs/user_guide/advanced_composition/nested_parallelization.md b/docs/user_guide/advanced_composition/nested_parallelization.md index c65987bd37..739335d4fc 100644 --- a/docs/user_guide/advanced_composition/nested_parallelization.md +++ b/docs/user_guide/advanced_composition/nested_parallelization.md @@ -17,7 +17,7 @@ kernelspec: +++ {"lines_to_next_cell": 0} -(map_task)= +(nested_parallelization)= # Nested Parallelization @@ -103,14 +103,28 @@ def multi_wf(l: typing.List[int], chunk: int) -> typing.List[int]: This shows a top-level workflow which uses 2 levels of dynamic workflows to process a list through some simple addition tasks and then flatten it again. Here is a visual representation of the execution in a Flyte console: -:::{figure} https://github.com/flyteorg/static-resources/blob/main/flytesnacks/user_guide/flyte_nested_parallelization.png?raw=true +:::{figure} https://raw.githubusercontent.com/flyteorg/static-resources/main/flytesnacks/user_guide/nested_parallel_top_level.png?raw=true :alt: Nested Parallelization UI View :class: with-shadow ::: +For each node in that top-level we can see 2 sub-workflows being run in parallel. + +:::{figure} https://raw.githubusercontent.com/flyteorg/static-resources/main/flytesnacks/user_guide/nested_parallel_inner_dynamic.png?raw=true +:alt: Inner Dynamic +:class: with-shadow +::: + +Finally, drilling into each sub-workflow, we'll see both those tasks being executed in series. + +:::{figure} https://raw.githubusercontent.com/flyteorg/static-resources/main/flytesnacks/user_guide/nested_parallel_subworkflow.png?raw=true +:alt: Sub-Workflow +:class: with-shadow +::: + ## Mixed Parallelism -This example is similar to the above but instead of using a dynamic to loop through a 2 task workflow in series, we're using that workflow to call a map task which processes both inputs in parallel. +This example is similar to the above but instead of using a dynamic to parallelize a 2-task serial workflow, we're using that workflow to call a map task which processes both inputs in parallel. This workflow has one fewer layers of parallelism so the outputs won't be the same, but it does still demonstrate how you can mix these different approaches to achieving concurrency. ```python """ @@ -160,11 +174,17 @@ def multi_wf(l: typing.List[int], chunk: int) -> typing.List[int]: ``` +++ {"lines_to_next_cell": 0} -## Concurrency vs Max Parallelism +While the top-level dynamic will be exactly the same as the previous example, here you can see the inner map task nodes as well as links in the sidebar. -Should we document this distinction or is it going away soon? +:::{figure} https://raw.githubusercontent.com/flyteorg/static-resources/main/flytesnacks/user_guide/nested_parallel_inner_map.png?raw=true +:alt: Inner Map Task +:class: with-shadow +::: ## Design Considerations You can nest even further if needed, or incorporate map tasks if your inputs are all the same type. The design of your workflow should of course be informed by the actual data you're processing. For example if you have a big library of music that you'd like to extract the lyrics for, the first level can loop through all the albums and the second level can go through each song. +If you're just churning through an enormous list of the same input, it's typically best to keep the code simple and let the scheduler handle optmizing the execution. Additionally, unless you need the features of a dynamic workflow like mixing and matching inputs and outputs, it's usually most efficient to use a map task. This has the added benefit of keeping the UI clean. + +You can also choose to limit the scale of parallel execution at a couple levels. The `max_parallelism` attribute can be applied at the workflow level and will limit the number of parallel tasks being executed, this is set to 25 by default. Within map tasks specifically, you can indicate a `concurrency` argument which will limit the number of mapped tasks within running at any given time. \ No newline at end of file From 6da6c72cda43cd76a7b534cf27c8faa23fc46307 Mon Sep 17 00:00:00 2001 From: pryce-turner Date: Fri, 8 Mar 2024 10:36:43 -0800 Subject: [PATCH 3/7] Spelling Signed-off-by: pryce-turner --- docs/user_guide/advanced_composition/nested_parallelization.md | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/docs/user_guide/advanced_composition/nested_parallelization.md b/docs/user_guide/advanced_composition/nested_parallelization.md index 739335d4fc..2e16638204 100644 --- a/docs/user_guide/advanced_composition/nested_parallelization.md +++ b/docs/user_guide/advanced_composition/nested_parallelization.md @@ -185,6 +185,6 @@ While the top-level dynamic will be exactly the same as the previous example, he You can nest even further if needed, or incorporate map tasks if your inputs are all the same type. The design of your workflow should of course be informed by the actual data you're processing. For example if you have a big library of music that you'd like to extract the lyrics for, the first level can loop through all the albums and the second level can go through each song. -If you're just churning through an enormous list of the same input, it's typically best to keep the code simple and let the scheduler handle optmizing the execution. Additionally, unless you need the features of a dynamic workflow like mixing and matching inputs and outputs, it's usually most efficient to use a map task. This has the added benefit of keeping the UI clean. +If you're just churning through an enormous list of the same input, it's typically best to keep the code simple and let the scheduler handle optimizing the execution. Additionally, unless you need the features of a dynamic workflow like mixing and matching inputs and outputs, it's usually most efficient to use a map task. This has the added benefit of keeping the UI clean. You can also choose to limit the scale of parallel execution at a couple levels. The `max_parallelism` attribute can be applied at the workflow level and will limit the number of parallel tasks being executed, this is set to 25 by default. Within map tasks specifically, you can indicate a `concurrency` argument which will limit the number of mapped tasks within running at any given time. \ No newline at end of file From 8ec5a0413aca73b136c958413a08c46ded220dd1 Mon Sep 17 00:00:00 2001 From: pryce-turner Date: Fri, 8 Mar 2024 10:51:46 -0800 Subject: [PATCH 4/7] toctree Signed-off-by: pryce-turner --- docs/user_guide/advanced_composition/index.md | 1 + 1 file changed, 1 insertion(+) diff --git a/docs/user_guide/advanced_composition/index.md b/docs/user_guide/advanced_composition/index.md index 26eb8df33c..3d0f14fc83 100644 --- a/docs/user_guide/advanced_composition/index.md +++ b/docs/user_guide/advanced_composition/index.md @@ -16,6 +16,7 @@ chaining_flyte_entities subworkflows dynamic_workflows map_tasks +nested_parallelization eager_workflows decorating_tasks decorating_workflows From 035e4ec4822f9b5c441b8ee4ac1518d9041dfe4b Mon Sep 17 00:00:00 2001 From: pryce-turner <31577879+pryce-turner@users.noreply.github.com> Date: Mon, 11 Mar 2024 11:16:17 -0700 Subject: [PATCH 5/7] Apply suggestions from code review Adding most suggestions with a few that need to be handled locally! Co-authored-by: Nikki Everett Signed-off-by: pryce-turner <31577879+pryce-turner@users.noreply.github.com> --- .../nested_parallelization.md | 49 ++++++++++--------- 1 file changed, 27 insertions(+), 22 deletions(-) diff --git a/docs/user_guide/advanced_composition/nested_parallelization.md b/docs/user_guide/advanced_composition/nested_parallelization.md index 2e16638204..c7602f3104 100644 --- a/docs/user_guide/advanced_composition/nested_parallelization.md +++ b/docs/user_guide/advanced_composition/nested_parallelization.md @@ -15,19 +15,18 @@ kernelspec: name: python3 --- -+++ {"lines_to_next_cell": 0} (nested_parallelization)= -# Nested Parallelization +# Nested parallelization ```{eval-rst} .. tags:: Advanced ``` -For exceptionally large or complicated workflows where dynamic workflows or map tasks aren't enough, it can be benficial to have multiple levels of parallelization. +For exceptionally large or complicated workflows that can't be adequately implemented as dynamic workflows or map tasks, it can be beneficial to have multiple levels of workflow parallelization. -This is useful for multiple reasons: +Nested parallelization enables: 1. Better code organization 2. Better code reuse 3. Better testing @@ -35,13 +34,15 @@ This is useful for multiple reasons: 5. Better monitoring - each subworkflow can be run independently and monitored independently 6. Better performance and scale - each subworkflow is executed as a separate workflow and thus can land on different flytepropeller workers and shards. This allows for better parallelism and scale. -## Nested Dynamics +## Nested dynamic workflows -This example shows how to break down a large workflow into smaller workflows and then compose them together to form a hierarchy. +You can use nested dynamic workflows to break down a large workflow into smaller workflows and then compose them together to form a hierarchy. In this example, a top-level workflow uses two levels of dynamic workflows to process a list through some simple addition tasks and then flatten the list again. + +### Example code ```python """ -The structure for 6 items and a chunk size of 2, the workflow will be broken down as follows: +A core workflow parallelized as six items with a chunk size of two will be structured as follows: multi_wf -> level1 -> level2 -> core_wf -> step1 -> step2 -> core_wf -> step1 -> step2 @@ -99,36 +100,39 @@ def multi_wf(l: typing.List[int], chunk: int) -> typing.List[int]: return level1(l=l, chunk=chunk) ``` -+++ {"lines_to_next_cell": 0} -This shows a top-level workflow which uses 2 levels of dynamic workflows to process a list through some simple addition tasks and then flatten it again. Here is a visual representation of the execution in a Flyte console: +### Flyte console + +Here is a visual representation of the execution of nested dynamic workflows in the Flyte console: :::{figure} https://raw.githubusercontent.com/flyteorg/static-resources/main/flytesnacks/user_guide/nested_parallel_top_level.png?raw=true :alt: Nested Parallelization UI View :class: with-shadow ::: -For each node in that top-level we can see 2 sub-workflows being run in parallel. +In each level2 node at the top level, two core workflows are run in parallel: :::{figure} https://raw.githubusercontent.com/flyteorg/static-resources/main/flytesnacks/user_guide/nested_parallel_inner_dynamic.png?raw=true -:alt: Inner Dynamic +:alt: Inner dynamic workflow :class: with-shadow ::: -Finally, drilling into each sub-workflow, we'll see both those tasks being executed in series. +Finally, in each core workflow, the two tasks are executed in series: :::{figure} https://raw.githubusercontent.com/flyteorg/static-resources/main/flytesnacks/user_guide/nested_parallel_subworkflow.png?raw=true -:alt: Sub-Workflow +:alt: Core workflow :class: with-shadow ::: -## Mixed Parallelism +## Mixed parallelism -This example is similar to the above but instead of using a dynamic to parallelize a 2-task serial workflow, we're using that workflow to call a map task which processes both inputs in parallel. This workflow has one fewer layers of parallelism so the outputs won't be the same, but it does still demonstrate how you can mix these different approaches to achieving concurrency. +This example is similar to nested dynamic workflows, but instead of using a dynamic workflow to parallelize a core workflow with serial tasks, we use a core workflow to call a map task, which processes both inputs in parallel. This workflow has one less layer of parallelism, so the outputs won't be the same as those of the nested parallelization example, but it does still demonstrate how you can mix these different approaches to achieve concurrency. + +### Example code ```python """ -The structure for 6 items and a chunk size of 2, the workflow will be broken down as follows: +A core workflow parallelized as six items with a chunk size of two will be structured as follows: multi_wf -> level1 -> level2 -> mappable -> mappable @@ -172,19 +176,20 @@ def multi_wf(l: typing.List[int], chunk: int) -> typing.List[int]: return level1(l=l, chunk=chunk) ``` -+++ {"lines_to_next_cell": 0} -While the top-level dynamic will be exactly the same as the previous example, here you can see the inner map task nodes as well as links in the sidebar. +### Flyte console + +While the top-level dynamic workflow will be exactly the same as the nested dynamic workflows example, the inner map task nodes will be visible as links in the sidebar: :::{figure} https://raw.githubusercontent.com/flyteorg/static-resources/main/flytesnacks/user_guide/nested_parallel_inner_map.png?raw=true :alt: Inner Map Task :class: with-shadow ::: -## Design Considerations +## Design considerations -You can nest even further if needed, or incorporate map tasks if your inputs are all the same type. The design of your workflow should of course be informed by the actual data you're processing. For example if you have a big library of music that you'd like to extract the lyrics for, the first level can loop through all the albums and the second level can go through each song. +While you can nest even further if needed, or incorporate map tasks if your inputs are all the same type, the design of your workflow should be informed by the actual data you're processing. For example, if you have a big library of music from which you'd like to extract the lyrics, the first level could loop through all the albums, and the second level could process each song. -If you're just churning through an enormous list of the same input, it's typically best to keep the code simple and let the scheduler handle optimizing the execution. Additionally, unless you need the features of a dynamic workflow like mixing and matching inputs and outputs, it's usually most efficient to use a map task. This has the added benefit of keeping the UI clean. +If you're just processing an enormous list of the same input, it's best to keep your code simple and let the scheduler handle optimizing the execution. Additionally, unless you need dynamic workflow features like mixing and matching inputs and outputs, it's usually most efficient to use a map task, which has the added benefit of keeping the UI clean. -You can also choose to limit the scale of parallel execution at a couple levels. The `max_parallelism` attribute can be applied at the workflow level and will limit the number of parallel tasks being executed, this is set to 25 by default. Within map tasks specifically, you can indicate a `concurrency` argument which will limit the number of mapped tasks within running at any given time. \ No newline at end of file +You can also choose to limit the scale of parallel execution at a few levels. The `max_parallelism` attribute can be applied at the workflow level and will limit the number of parallel tasks being executed. (This is set to 25 by default.) Within map tasks, you can specify a `concurrency` argument, which will limit the number of mapped tasks that can run in parallel at any given time. \ No newline at end of file From f6a27be9411a67d361f65c22c7819109942610af Mon Sep 17 00:00:00 2001 From: pryce-turner Date: Mon, 11 Mar 2024 11:19:22 -0700 Subject: [PATCH 6/7] Merge --- .../nested_parallelization.md | 32 ++++--------------- 1 file changed, 7 insertions(+), 25 deletions(-) diff --git a/docs/user_guide/advanced_composition/nested_parallelization.md b/docs/user_guide/advanced_composition/nested_parallelization.md index c7602f3104..96f63fdab7 100644 --- a/docs/user_guide/advanced_composition/nested_parallelization.md +++ b/docs/user_guide/advanced_composition/nested_parallelization.md @@ -1,21 +1,3 @@ ---- -jupytext: - cell_metadata_filter: all - formats: md:myst - main_language: python - notebook_metadata_filter: all - text_representation: - extension: .md - format_name: myst - format_version: 0.13 - jupytext_version: 1.16.1 -kernelspec: - display_name: Python 3 - language: python - name: python3 ---- - - (nested_parallelization)= # Nested parallelization @@ -26,13 +8,13 @@ kernelspec: For exceptionally large or complicated workflows that can't be adequately implemented as dynamic workflows or map tasks, it can be beneficial to have multiple levels of workflow parallelization. -Nested parallelization enables: -1. Better code organization -2. Better code reuse -3. Better testing -4. Better debugging -5. Better monitoring - each subworkflow can be run independently and monitored independently -6. Better performance and scale - each subworkflow is executed as a separate workflow and thus can land on different flytepropeller workers and shards. This allows for better parallelism and scale. +This is useful for multiple reasons: +- Better code organization +- Better code reuse +- Better testing +- Better debugging +- Better monitoring, since each subworkflow can be run independently and monitored independently +- Better performance and scale, since each subworkflow is executed as a separate workflow and thus can be distributed among different flytepropeller workers and shards. This allows for better parallelism and scale. ## Nested dynamic workflows From b2398c891b986ff0086b33be93f921b028a97df4 Mon Sep 17 00:00:00 2001 From: pryce-turner Date: Mon, 11 Mar 2024 12:05:42 -0700 Subject: [PATCH 7/7] Updated screenshot with annotation Signed-off-by: pryce-turner --- docs/user_guide/advanced_composition/nested_parallelization.md | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/docs/user_guide/advanced_composition/nested_parallelization.md b/docs/user_guide/advanced_composition/nested_parallelization.md index 96f63fdab7..73fd921b2e 100644 --- a/docs/user_guide/advanced_composition/nested_parallelization.md +++ b/docs/user_guide/advanced_composition/nested_parallelization.md @@ -94,7 +94,7 @@ Here is a visual representation of the execution of nested dynamic workflows in In each level2 node at the top level, two core workflows are run in parallel: -:::{figure} https://raw.githubusercontent.com/flyteorg/static-resources/main/flytesnacks/user_guide/nested_parallel_inner_dynamic.png?raw=true +:::{figure} https://raw.githubusercontent.com/flyteorg/static-resources/main/flytesnacks/user_guide/nested_parallel_inner_dynamic_anno.png?raw=true :alt: Inner dynamic workflow :class: with-shadow :::