Add Parallel interface for executing recipes in parallel #138

rushirajnenuji · 2025-12-18T20:20:25Z

This PR adds parallel task execution to ogdc-runner, allowing workflows to process large datasets faster by running tasks concurrently in Argo.

rel: #136

Add core parallel execution model

…models Add parallel execution config field to ShellWorkflow and VizWorkflow models

Rename module to parallel_config

Update parallel_config.py

[WIP] Add partitioning and parallel exec modules

Align parallel execution with parallel_config model

Remove feature based batching for viz-workflow

Add support for rasterization and 3dtiles initial work on this PR #114

…allel-enhancement

…enhancement

Utilize parallel orchestration for shell type workflows

resolve issues with circular imports

[WIP] fix issues with adding exec steps to a DAG marks first successful test for running the shell workflow in parallel.

[WIP] clean up

Decided to handle these changes in a new branch

[WIP] update parallel partitioning approach and clean up

Removing filesystem type from recipe inputs Within K8s, I wonder if we're ever going to access the file paths directly from local system. I think the K8s way is to access the object via PVC

[WIP] use PVC instead of artifacts for parallel execution

[WIP] Clean up and refactor parallel execution logic

Fix PVC mount name and file iteration

Make max parallel limit configurable via env-var

Update documentation

Fix issues with mypy

trey-stafford

I realize this is still in draft form, but wanted to get some initial feedback in before the holidays.

Lot of good work here - nice job! I do have some suggested changes and raise a couple of questions that we may want to discuss further. Overall though this is a great feature to have!

Do you have any "real-world" recipes for this feature drafted? It would be nice to see how e.g,. the PDG plans to use this. Maybe we could use that as an example?

trey-stafford · 2025-12-23T17:25:13Z

tests/test_parallel_shell_recipe_dir/recipe.sh

+# Each command will be executed in parallel across partitions
+
+# Step 1: Process files - add header and line numbers
+cat "$INPUT_FILE" | awk 'BEGIN {print "--- Processed File ---"} {print NR": "$0}' > "$OUTPUT_FILE"


This does not follow the shell recipe convention of using /input_dir/ and /output_dir/ to represent the input/output directory for each step.

E.g., see: https://ogdc-runner.readthedocs.io/en/latest/recipes.html#shell-workflow

It is expected that each command in the recipe.sh place data in /output_dir/

The input data for each step is always assumed to be in /input_dir/. The previous step’s /output_dir/ becomes the next step’s /input_dir/. The first step’s /input_dir/ contains the data specified in the meta.yaml’s input.

Each command is executed in isolation. Do not expect envvars (e.g., export ENVVAR=foo) to persist between lines.

We should try to make this consistent across sequential and parallel shell recipes. Maybe we could consolidate around $INPUT and $OUTPUT? Alternatively, this might be a reason to separate the parallel shell recipe into it's own type (parallel-shell), to clearly distinguish it from the sequential case.

Still think we need to address this. At a minimum, let's create an issue to return to once this PR is complete.

docs/architecture/index.md

src/ogdc_runner/inputs.py

trey-stafford · 2025-12-23T17:37:55Z

src/ogdc_runner/models/parallel_config.py

+    @field_validator("function", mode="before")
+    @classmethod
+    def validate_function(cls, v: Any) -> Any:
+        """Validate that function is callable if provided.


Does pydantic not automatically validate that this is a callable based on the typing above? Or does exclude prevent that? What's the use-case here?

Thoughts on this?

src/ogdc_runner/parallel.py

trey-stafford · 2025-12-23T17:52:13Z

src/ogdc_runner/parallel.py

+        msg = f"ExecutionFunction '{func.name}' must have 'command' or 'function'"
+        raise ValueError(msg)
+
+    def _create_shell_template(self, func: ExecutionFunction) -> Container:


Consider moving this (and other shell-specific) logic to the shell module? I would strive to make this module (and the other partition/parallelization-related modules) to define high-level abstractions that can be used by recipe-specific implementations. his could look like defining an abstract class here, and then creating a ShellParallelExecutionOrchestrator class specifically for shell workflows that inherits from that.

That said, this might make more sense to tackle as part of supporting parallelization for viz workflows, or in another follow-on PR.

Thoughts on this?

tests/unit/test_parallel_recipe.py

trey-stafford · 2025-12-23T18:18:27Z

docs/recipes.md

+
+Each parallel task:
+
+- Receives a partition of input files via workflow parameters


Is it expected that each task have the same number of files?

trey-stafford · 2025-12-23T18:19:22Z

docs/recipes.md

+Each parallel task:
+
+- Receives a partition of input files via workflow parameters
+- Executes the same command/function independently


Does this mean that each command will be run once per file? Or do the underlying commands need to be capable of handling the partition of files passed to them?

Fix inputs for subsequent shell cmds

fix mypy errors

Add more fixes

remove get_max_parallelism

update documentation

Move test_parallel_recipe to unit test

Update recipe

Resolve type checking issues

remove version specific decorators

trey-stafford

Getting close! Nice work on this - it's a big change!

trey-stafford · 2026-01-22T17:00:07Z

docs/architecture/index.md

+2. **Creates** independent Argo tasks for each partition
+3. **Orchestrates** parallel execution with configurable maximum parallelism
+
+The `ParallelExecutionOrchestrator` class manages this process, creating Argo


Suggested change

The `ParallelExecutionOrchestrator` class manages this process, creating Argo

The {class}`ogdc_runner.parallel.ParallelExecutionOrchestrator` class manages this process, creating Argo

trey-stafford · 2026-01-22T17:00:25Z

docs/architecture/index.md

+- {mod} `ogdc_runner.parallel`: Orchestration logic for parallel task creation
+- {mod} `ogdc_runner.partitioning`: Partitioning strategies for dividing work
+- {mod} `ogdc_runner.models.parallel_config`: Configuration models for parallel


Suggested change

- {mod} `ogdc_runner.parallel`: Orchestration logic for parallel task creation

- {mod} `ogdc_runner.partitioning`: Partitioning strategies for dividing work

- {mod} `ogdc_runner.models.parallel_config`: Configuration models for parallel

- {mod}`ogdc_runner.parallel`: Orchestration logic for parallel task creation

- {mod}`ogdc_runner.partitioning`: Partitioning strategies for dividing work

- {mod}`ogdc_runner.models.parallel_config`: Configuration models for parallel

trey-stafford · 2026-01-22T17:13:51Z

src/ogdc_runner/inputs.py

+    """
+    if use_input_as_output:
+
+        def get_recipe_inputs_path(recipe_id: str) -> str:


This function appears to be unused and should be removed.

trey-stafford · 2026-01-22T17:16:00Z

src/ogdc_runner/inputs.py

+            artifact storage.
+
+    Returns:
+        Output directory path


Suggested change

Output directory path

Output directory as a string.

trey-stafford · 2026-01-22T17:18:26Z

src/ogdc_runner/inputs.py

+        use_input_as_output: If True, store inputs in PVC at a path that can be
+            referenced as output for downstream recipes. If False, use temporary
+            artifact storage.


Suggested change

use_input_as_output: If True, store inputs in PVC at a path that can be

referenced as output for downstream recipes. If False, use temporary

artifact storage.

use_input_as_output: If True, return `"/mnt/workflow/{recipe_id}/inputs"`. Otherwise `/output_dir`.

It is not accurate to say that this function stores inputs on a PVC or artifact storage. It just returns a str.

trey-stafford · 2026-01-22T17:24:13Z

src/ogdc_runner/parallel.py

+            Complete shell script as a string
+        """
+        return f"""
+set -e


Maybe we could move this string into a shell file that gets read here instead of inlining it? This would make it easier for editors to pick up on shell syntax issues and generally make it easier to read/edit.

trey-stafford · 2026-01-22T17:27:09Z

src/ogdc_runner/publish.py

+    command = "rsync --progress /input_dir/* /output_dir/"
+    volume_mounts = [
+        models.VolumeMount(
+            name=OGDC_WORKFLOW_PVC.name, mount_path="/output_dir/", sub_path=recipe_id


Suggested change

name=OGDC_WORKFLOW_PVC.name, mount_path="/output_dir/", sub_path=recipe_id

name=OGDC_WORKFLOW_PVC.name,

mount_path="/output_dir/",

sub_path=recipe_id,

Bit of a nitpick here: easier to read kwargs passed when they are each on a separate line (and easier to add new kwargs w/o going over line-length).

trey-stafford · 2026-01-22T17:34:26Z

tests/test_parallel_shell_recipe_dir/recipe.sh

+# Each command will be executed in parallel across partitions
+
+# Step 1: Process files - add header and line numbers
+cat "$INPUT_FILE" | awk 'BEGIN {print "--- Processed File ---"} {print NR": "$0}' > "$OUTPUT_FILE"


Still think we need to address this. At a minimum, let's create an issue to return to once this PR is complete.

trey-stafford · 2026-01-22T17:34:39Z

src/ogdc_runner/models/parallel_config.py

+    @field_validator("function", mode="before")
+    @classmethod
+    def validate_function(cls, v: Any) -> Any:
+        """Validate that function is callable if provided.


Thoughts on this?

trey-stafford · 2026-01-22T17:34:59Z

src/ogdc_runner/parallel.py

+        msg = f"ExecutionFunction '{func.name}' must have 'command' or 'function'"
+        raise ValueError(msg)
+
+    def _create_shell_template(self, func: ExecutionFunction) -> Container:


Thoughts on this?

Feedback suggestions and clean up

Move partition script to a separate module

refactor: abstract ParallelExecutionOrchestrator - abstract ParallelExecutionOrchestrator module defines the interface - move shell specific logic to shell module

Fix shellcheck issues and updated documentation

Fix failing test

rushirajnenuji added 28 commits November 20, 2025 11:36

Add core parallel execution model

1ad2684

Add core parallel execution model

Add parallel execution config field to ShellWorkflow and VizWorkflow …

a70581a

…models Add parallel execution config field to ShellWorkflow and VizWorkflow models

Rename module to parallel_config

0bb1d54

Rename module to parallel_config

Update parallel_config.py

38ff544

Update parallel_config.py

[WIP] Add partitioning and parallel exec modules

074b555

[WIP] Add partitioning and parallel exec modules

Align parallel execution with parallel_config model

46b03d4

Align parallel execution with parallel_config model

Remove feature based batching for viz-workflow

bd106be

Remove feature based batching for viz-workflow

Add support for rasterization and 3dtiles

da72fb9

Add support for rasterization and 3dtiles initial work on this PR #114

Merge branch '101-fix-viz-workflow-remote-recipes' into feat-ogdc-par…

cce1874

…allel-enhancement

Merge branch 'validator-for-config-filepath' into feat-ogdc-parallel-…

0534315

…enhancement

Utilize parallel orchestration for shell type workflows

1595262

Utilize parallel orchestration for shell type workflows

Resolve issues with circular imports

b2410fa

resolve issues with circular imports

[WIP] fix issues with adding exec steps to a DAG

b0e2b18

[WIP] fix issues with adding exec steps to a DAG marks first successful test for running the shell workflow in parallel.

[WIP] clean up

27d70d8

[WIP] clean up

Merge branch 'main' into feat-ogdc-parallel-enhancement

c9aa56e

Merge branch 'main' into feat-ogdc-parallel-enhancement

215e465

revert viz-wokflow.py

38a135f

Decided to handle these changes in a new branch

[WIP] update parallel partitioning approach and clean up

13bf9ad

[WIP] update parallel partitioning approach and clean up

Removing filesystem type from recipe inputs

a0f9800

Removing filesystem type from recipe inputs Within K8s, I wonder if we're ever going to access the file paths directly from local system. I think the K8s way is to access the object via PVC

[WIP] use PVC instead of artifacts for parallel execution

a67a795

[WIP] use PVC instead of artifacts for parallel execution

[WIP] Clean up and refactor parallel execution logic

c6c04c1

[WIP] Clean up and refactor parallel execution logic

Fix PVC mount name and file iteration

dcaa9b8

Fix PVC mount name and file iteration

Make max parallel limit configurable via env-var

3b73bdd

Make max parallel limit configurable via env-var

[WIP] tests

b0577bf

Update documentation

a5ebaeb

Update documentation

Fix parallel test

1741617

Fix issues with mypy

6769bd4

Fix issues with mypy

Merge branch 'main' into feat-ogdc-parallel-enhancement

39dca16

trey-stafford requested changes Dec 23, 2025

View reviewed changes

Merge branch 'main' into feat-ogdc-parallel-enhancement

b989d10

rushirajnenuji added 8 commits January 13, 2026 11:20

Fix inputs for subsequent shell cmds

a46372e

Fix inputs for subsequent shell cmds

fix mypy errors

e8e950a

fix mypy errors

Add more fixes

8517939

Add more fixes

remove get_max_parallelism

5f551a7

remove get_max_parallelism

update documentation

f50f679

update documentation

Merge branch 'main' into feat-ogdc-parallel-enhancement

1287ddb

Move test_parallel_recipe to unit test

c125180

Move test_parallel_recipe to unit test

Update recipe

2ea0152

Update recipe

rushirajnenuji marked this pull request as ready for review January 14, 2026 16:56

rushirajnenuji added 2 commits January 14, 2026 09:08

Resolve type checking issues

d750734

Resolve type checking issues

remove version specific decorators

bda31e8

remove version specific decorators

rushirajnenuji requested review from rmarow and trey-stafford January 21, 2026 19:10

trey-stafford requested changes Jan 22, 2026

View reviewed changes

rushirajnenuji added 6 commits February 3, 2026 18:09

Incorporated feedback suggestions and clean up

cfbe601

Feedback suggestions and clean up

Move partition script to a separate module

874387e

Move partition script to a separate module

refactor: abstract ParallelExecutionOrchestrator

179331f

refactor: abstract ParallelExecutionOrchestrator - abstract ParallelExecutionOrchestrator module defines the interface - move shell specific logic to shell module

Fix shellcheck issues and updated documentation

0a8bfbc

Fix shellcheck issues and updated documentation

Merge branch 'main' into feat-ogdc-parallel-enhancement

660b166

Fix failing test

cf8ce06

Fix failing test

rushirajnenuji requested a review from trey-stafford February 4, 2026 17:14


		Each parallel task:

		- Receives a partition of input files via workflow parameters

	The `ParallelExecutionOrchestrator` class manages this process, creating Argo
	The {class}`ogdc_runner.parallel.ParallelExecutionOrchestrator` class manages this process, creating Argo

Add Parallel interface for executing recipes in parallel #138

Are you sure you want to change the base?

Add Parallel interface for executing recipes in parallel #138

Uh oh!

Conversation

rushirajnenuji commented Dec 18, 2025

Uh oh!

trey-stafford left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

trey-stafford left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

2 participants