Fan-out, Map and Gather

What this workflow does

This workflow demonstrates the declarative map: construct of horus-runtime.

A producer stage splits a small input collection into batches. A map: block fans one stage out over every batch. The clones run concurrently. A gather stage collects the N results back into a single folder.

There is no science here. Every stage shells out to echo, wc, or tr, or it runs a stdlib-only script. The workflow runs anywhere in a few seconds. Read it as a template for a real pipeline, such as a batched ligand library.

The problem it solves

A screen over a large collection is embarrassingly parallel. The items do not depend on each other. A serial stage still processes them one at a time.

To run them in parallel by hand, you write a loop, a job array, and a merge script. You then track which clone failed and which output belongs to which batch. That code is not part of your science, and you rewrite it per pipeline.

The map: block moves that work into the engine. You declare one stage and the collection it runs over. Horus creates one clone per element and gives each clone its own working directory. Horus places the clone outputs under a single <stage>.gathered/<i>/ tree.

The fan-in target is stable. The gather stage reads one folder and does not need to know the clone count. The clone count follows the data, so a bigger input adds clones without an edit to the graph.

Pipeline

split (local)             examples/items.json (8 strings) ──► batches/ (4 files)
   │  chunks the JSON list into batch_0..batch_3.txt, 2 items each

score[00..03] (map, 4x concurrent clones over batches/)
   │  each clone: word-counts + uppercases its one batch file
   │  ──► scored/count.txt, scored/upper.txt   (fan-in target: score.gathered/<i>/)

gather (local)             score.gathered/ (4 subfolders) ──► results/summary/
   │  sums the per-batch word counts, concatenates the uppercased text

Inputs and outputs

Input

• examples/items.json: a JSON array of 8 short strings.

Outputs land in workflow_results/:

• score.gathered/<i>/scored/count.txt: the word count for batch <i>. The bundled example gives 2 for every clone.
• score.gathered/<i>/scored/upper.txt: the batch text, uppercased.
• results/summary/summary.json: {"total_words": 8, "batches": [...]} for the bundled example.
• results/summary/combined_upper.txt: every batch text, concatenated in batch order.

Run the workflow

Install uv if you do not have it:

curl -LsSf https://astral.sh/uv/install.sh | sh

Then install the horus-runtime and run the workflow:

cd workflows/engine-showcases/w01-fanout-map-gather
uv sync
horus run workflow.yaml

You can also install the runtime with pip:

pip install horus-runtime

To change the fan-out width, edit the --batch-size argument of the split stage in workflow.yaml. The batch count sets the clone count. To use your own data, replace the array in examples/items.json.

References

• Horus repositories