VineReduce

A dynamic MapReduce framework for data processing, built on top of TaskVine.

For each (processor, dataset) pair, VineReduce splits every file in the dataset into chunks, runs your processor over each chunk remotely (the "map" step), then repeatedly folds pooled processor outputs together with a reducer (the "reduce" step) until one final result covers the whole dataset. Progress is checkpointed along the way, so an interrupted run can resume without redoing finished work. See PLAN.md for the full design.

Installation

This project requires Python 3.13+ and is managed with pixi.

# Clone the repository
git clone https://github.com/cooperative-computing-lab/vine-reduce.git
cd vine-reduce

# Install the default environment (runtime dependencies only)
pixi install

# Or install the dev environment (adds pytest, black, flake8, pyright)
pixi install -e dev

All commands should be run through pixi so they pick up the managed environment, e.g. pixi run python your_script.py.

Quick Start

examples/quick_start/quick_start.py is a self-contained, runnable example: it generates some toy binary data, starts a TaskVineDistributor with a single local worker (no cluster or separate vine_worker process needed), runs three processors over two datasets, and checks the results for consistency.

cd examples/quick_start
pixi run python quick_start.py

Reading through that file top-to-bottom (it's heavily commented) is the fastest way to see how the pieces fit together: build_datasets describes the input shape VineReduce expects, numbers_chunk_to_args turns a Chunk into processor arguments, and main() wires a TaskVineDistributor and VineReduce together and calls compute().

The shape of a minimal call looks like this:

from vine_reduce import VineReduce

vr = VineReduce(
    processors={"my_processor": my_processor_fn},
    input=datasets,  # {name: {"metadata": {...}, "files": {path: num_entries}}}
    chunk_to_args=my_chunk_to_args,
    chunksize=10_000,
    results_dir="results",
    # distributor defaults to a local ProcessPoolExecutor-backed
    # LocalDistributor if omitted; pass a TaskVineDistributor to run on a
    # real TaskVine cluster instead. Non-final checkpoints are the
    # distributor's own concern (e.g. TaskVineDistributor(checkpoint_dir=...)),
    # not VineReduce's.
)
vr.compute()

Executors

chunk_to_args's output for a chunk becomes the argument each processor call runs remotely on. The executor argument to VineReduce takes a constructed Executor instance — mirroring how distributor takes a constructed Distributor instance — that controls how that call actually runs, at the execution site. It's configured once, in the local process, and cloudpickled into every remote call, where defaults.executor_wrapper uses it as:

executor.submit(
    processor, args,
    dataset_metadata=..., distributor_metadata=..., executor_metadata=...,
).result()

Executor follows concurrent.futures.Executor's submit/map/shutdown shape (and is usable as a context manager), though submit/map take the three metadata dicts above as extra keyword arguments.

• SimpleExecutor() (default) — calls processor(args) directly.
• CloudpickleExecutor(max_workers=1) — runs processor(args) in its own subprocess, so a crash or memory leak in processor doesn't take down the worker task itself. Supports closures and lambdas as processor, unlike the stdlib pickle a plain ProcessPoolExecutor would require. A max_workers above 1 lets map() run items in parallel.
• DaskExecutor(num_workers=None) — for a processor that returns a dask-delayed object (or dask array/dataframe) rather than a plain value; computes it at the execution site using num_workers subprocesses, or (if not given) one subprocess per core allocated to the task. dask is not a vine_reduce dependency and must already be installed wherever this executor runs.

All three live in src/vine_reduce/executor.py.

Packaging an environment for remote workers

TaskVineDistributor accepts an environment= argument — a path to a packed, relocatable poncho package tarball — which it ships to every worker alongside each task, so worker nodes need nothing beyond TaskVine itself pre-installed.

vine_reduce.get_environment() (src/vine_reduce/remote_environment.py) builds that tarball for you, via poncho_package_create:

from vine_reduce import TaskVineDistributor, get_environment

environment = get_environment()
distributor = TaskVineDistributor(
    port=0,
    resources_processor={"cores": 1},
    environment=environment,
)

get_environment() packs whatever is currently installed in the calling conda environment ($CONDA_PREFIX by default, or pass conda_env_path=) - nothing more. Install whatever your workers need (conda install, pip install, a pixi dependency, ...) into that environment before calling it. Builds are cached on disk (keyed by a hash of the environment's installed packages) and reused across runs. Editable pip installs can't be packed as-is, so any package currently installed editable (vine_reduce itself, typically) is temporarily reinstalled non-editable for the pack step and reinstalled editable again immediately afterwards. If vine_reduce - or another package named via pip_editable - has uncommitted changes in its checkout, the next call rebuilds automatically rather than risk shipping stale code (pass unstaged="fail" to raise UnstagedChanges instead):

environment = get_environment(
    pip_editable={"my-analysis-repo": ["src", "pyproject.toml"]},
)

get_environment is not TaskVine-specific: it just resolves a tarball path, so it works the same way regardless of which Distributor ultimately uses that path. Building requires poncho_package_create and conda on PATH - see the conda extra in pyproject.toml.

HEP / coffea workflows

vine_reduce.VineReduceCoffea is a specialization for coffea-based analyses: it supplies NanoEvents-reading, awkward-array materialization, and coffea-style accumulator merging, while chunking, checkpointing, and restart are inherited unchanged from VineReduce. See src/vine_reduce/coffea.py.

examples/cortado/vr_cortado.py is a runnable example built on it, adapted from the "cortado" example in dynamic_data_reduction, the project this one's dynamic map-reduce loop descends from: it generates synthetic NanoAOD-like ROOT files for two datasets, skims each down to events with at least four leptons, and merges the surviving events per dataset with a custom awkward-array-concatenating reducer.

cd examples/cortado
pixi run python vr_cortado.py

Production use: ttbarEFT

TopEFT/ttbarEFT is a CMS top-quark EFT search that runs its analysis stage through vine_reduce on top of TaskVine, distributing histogram-filling processors over an HTCondor pool. examples/ttBar/run_processor_with_vr.py shows how that integration looked in practice: driving a ttbarEFT AnalysisProcessor per lepton channel through vine_reduce. It predates the current VineReduceCoffea/TaskVineDistributor API described above (it was written against an earlier vine_reduce release), so treat it as a reference for how a full physics analysis wires up channels, Wilson-coefficient/histogram selection, and X509 proxy handling around vine_reduce, not as a runnable script against the current API.

Development

pixi run -e dev pytest tests/ -v   # run tests
pixi run -e dev black .            # format
pixi run -e dev flake8             # lint

CI (GitHub Actions and a mirrored GitLab CI pipeline) runs all three on every push and pull request.

License

This project is licensed under the Apache License 2.0 - see the LICENSE file for details.