coffea

Coffea uses a columnar data model (via Awkward Array) paired with a thin execution layer. This allows analysis code to transition from a local laptop to a high-performance cluster without code rewrites. The standard workflow follows three steps:

1. Implement a Processor: Create a class inheriting from coffea.processor.ProcessorABC that transforms input data (e.g., NanoEvents) into accumulators (like histograms).
2. Iterate Locally: Execute the processor using coffea.processor.Runner with a local executor (like IterativeExecutor) to debug and refine the logic.
3. Scale Out: Swap the local executor for a distributed one (like DaskExecutor) when ready to process large datasets on a cluster.

import awkward as ak
import correctionlib
import hist
from coffea import processor

class MuonProcessor(processor.ProcessorABC):
    def __init__(self, sf_path: str):
        self.corrections = correctionlib.CorrectionSet.from_file(sf_path)
        self.muon_sf = self.corrections["muon_sf"]

    def process(self, events):
        # ... logic to transform events into accumulators ...
        return {"dataset": {"mass": h_mass, "events": len(events)}}

    def postprocess(self, accumulator):
        return accumulator

Coffea provides multiple executors that implement a unified interface, allowing you to switch from local debugging to large-scale cluster execution by only changing the executor argument in the Runner.

Local Executors

Used for running on a single machine:

• IterativeExecutor: Processes chunks sequentially in a single Python thread. Best for debugging and validation.
• FuturesExecutor: Uses concurrent.futures to parallelize work across multiple local workers (typically a process pool). You can tune parallelism using the workers or pool arguments.

Distributed Executors

Used for connecting to clusters and grid-style resources:

• DaskExecutor: Integrates with a Dask Distributed cluster via a distributed.Client.
• ParslExecutor: Uses Parsl to target various HPC and batch backends.
• TaskVineExecutor: Leverages TaskVine for opportunistic and heterogeneous workers.

Coffea uses a pattern where each chunk of rows is processed independently, allowing for easy horizontal scale-out.

• coffea.processor.ProcessorABC: You implement your analysis by subclassing this class and defining a process method that returns an accumulator.
• coffea.processor.Runner: This helper manages the orchestration. It handles dataset chunking, the creation of NanoEvents, and the reduction of per-chunk results into a final output.

The Runner allows you to execute your processor using the same interface regardless of whether you are running locally or on a distributed cluster.

from coffea import processor
from coffea.nanoevents import NanoAODSchema

# Assume ``my_processor`` is an instance of a subclass of ProcessorABC.
fileset = {
    "ZJets": {"treename": "Events", "files": ["/data/nano_dy.root"]},
    "Data": {"treename": "Events", "files": ["/data/nano_dimuon.root"]},
}

runner = processor.Runner(
    executor=processor.FuturesExecutor(workers=4, status=True),
    schema=NanoAODSchema,
)

result = runner(
    fileset,
    processor_instance=my_processor,
    treename="Events",
)

To contribute to Coffea, clone the repository and install it in editable mode with development dependencies.

Requirements:

• To use the --group flag with pip, you need pip version 25.1 or newer.
• Alternatively, you can use uv for environment management.

Steps:

1. Clone the source.
2. Install with the dev dependency group.
3. (Optional) Install with dask support if working on executors.
4. Run linting, tests, and documentation builds using pre-commit, pytest, and make.

# 1. Download source
git clone https://github.com/scikit-hep/coffea

# 2. Install with development dependencies
cd coffea
pip install --editable . --group dev

# Or if you need to work on the executors (e.g. dask)
pip install --editable '.[dask]' --group dev

# Or using uv
uv sync

# 4. Lint, test, and build docs
pre-commit run --all-files
pytest tests
pushd docs && make html && popd