Collecting and aggregation

Collectors reduce a flow into a result. agg supplies factories for common named and grouped aggregations; each result is also a Collector.

Since 2.4.0, named aggregate() and Pairs.aggregate_values() accept any Collector. This includes compositions made with Collectors.filtering, mapping, teeing, and collecting_and_then. The same collector can run over a Flow, an AsyncFlow, or each row group, including sorted and spill groups. Every group receives fresh state from its initializer.

aggregate() requires named collectors. For a single unnamed result, use collect(collector). Collector callbacks keep their usual synchronous contract when used with AsyncFlow. Composing a collector does not grant native execution or reducer laws; those still require their existing proofs and declarations. See the weighted collector recipe.

grouping_by groups all matching keys, even when they are not adjacent; group_runs instead emits consecutive runs. The grouping classifier receives each item, including None, and None itself can also be a group key. A group's initializer runs when its key first appears; a state of None is a valid state, not a request to initialize it again. partitioning_by initializes and finishes both Boolean partitions, including for empty input.

reduce_by also initializes once per encountered key, even when the initializer and subsequent steps return None. If a callback needs to tell the first item apart from a legitimate None state, use a distinct marker:

from fpstreams import flow

items = [("a", None), ("a", 7)]
initialized = []

def initial_none():
    initialized.append(True)
    return None

assert flow(items).reduce_by(
    lambda pair: pair[0], lambda state, pair: None, initializer=initial_none
) == {"a": None}
assert initialized == [True]

unseen = object()
assert flow(items).reduce_by(
    lambda pair: pair[0],
    lambda state, pair: pair[1] if state is unseen else state,
    initializer=lambda: unseen,
) == {"a": None}

The same per-key rule applies to await aflow(items).reduce_by(...); its selector, initializer, and step can also return awaitables.

grouping_by and partitioning_by offer every selected item to their downstream collector even if its done predicate is true. teeing stops stepping a finished downstream but stops the source only when both downstream collectors are done. Choose downstream callbacks that support the lifecycle of the chosen adapter. For AsyncFlow, collector callbacks remain synchronous: a classifier's returned Future is a key, whereas AsyncFlow.reduce_by resolves an awaitable selector result.

chunk and group_runs return tuples containing the original item references; they do not deep-copy mutable items. Arrow batch slices may share the source value buffer. Retaining a slice can therefore retain that buffer, and a row-count limit alone is not a byte-memory limit.

Collectors.frequencies() (2.5.0) returns a dictionary counting each hashable input, in first-encounter order. It uses one entry per distinct value and can be combined with mapping for grouped field counts:

from fpstreams import Collectors, flow

assert flow(["left", "right", "left"]).collect(Collectors.frequencies()) == {
    "left": 2, "right": 1
}

For example, count positions separately for each team:

from fpstreams import Collectors, rows

positions = Collectors.mapping("position", Collectors.frequencies())
result = rows([
    {"team": "a", "position": "left"},
    {"team": "a", "position": "right"},
    {"team": "a", "position": "left"},
    {"team": "b", "position": "right"},
    {"team": "b", "position": "left"},
]).group_by("team").aggregate(counts=positions).to_list()

assert [group["team"] for group in result] == ["a", "b"]
assert list(result[0]["counts"].items()) == [("left", 2), ("right", 1)]
assert list(result[1]["counts"].items()) == [("right", 1), ("left", 1)]

Collectors.mode() (2.5.0) returns the most frequent hashable value; ties go to the value first encountered. Empty input returns None, which is also a valid input value. The collector retains one entry per distinct value and does not provide a partial-state combiner. Compose it with mapping and filtering to select a field or exclude missing values:

from fpstreams import Collectors, rows

positions = Collectors.mapping("position", Collectors.mode())
result = rows([
    {"team": "a", "position": "left"},
    {"team": "a", "position": "right"},
    {"team": "a", "position": "right"},
]).group_by("team").aggregate(position=positions).to_list()
assert result == [{"team": "a", "position": "right"}]

Since 2.4.0, editor hints and type checkers retain the finished result type of a single streaming collector for both Flow and AsyncFlow. Named collector calls return dict[str, Any]. Calls that forward a possibly None collector together with named arguments keep the general result union.

AsyncFlow uses a streaming Collector's synchronous state-machine contract. An awaitable returned by its finisher remains a result value for the caller to await. For a non-Collector callable, AsyncFlow awaits an awaitable result before returning it. These runtime rules are unchanged by the type declarations.

Collector type inference (2.5.0)

Container factories infer their element type from the consuming pipeline or an explicit annotation. Collectors[int] remains available when constructing a collector separately:

from typing import assert_type

from fpstreams import Collector, Collectors, flow

assert_type(flow([1, 2]).collect(Collectors.to_list()), list[int])
assert_type(flow([1, 2]).collect(Collectors.to_tuple()), tuple[int, ...])
integers: Collector[int, list[int], list[int]] = Collectors[int].to_list()

Unannotated standalone factories and some inline Collectors[Any] calls remain checker dependent. Use a concrete Collectors[T] namespace or store the result with an explicit Collector annotation before consuming it.

The same result inference applies to AsyncFlow.collect() and Pairs.collect_values(). A collector consumes its input type; its mutable state and result types remain invariant. In particular, an integer list collector does not become a string list collector. An awaitable finisher remains a value in both the synchronous and asynchronous Collector contracts described above.

A separately annotated Collector[Any, list[Any], list[Any]] retains its list[Any] result in both Flow and AsyncFlow. Forwarding a possibly None collector still retains the result union with dict[str, Any] for named collection.

For an async Collector with an awaitable finisher, use a concrete input type; object is suitable if its step accepts arbitrary objects. Combining Any with the callable overloads can still erase this result type. The runtime continues to preserve the finisher's awaitable as a value.

Source installations use typing-extensions>=4.12 on Python 3.11 and 3.12 for type parameter defaults; Python 3.13 and newer use the standard library. This typing change is not included in the published 2.4.1 release.

Collector

fpstreams.Collector `dataclass`

Bases: Generic[T, S, R]

Describe an immutable streaming reduction from input items to a result.

initializer creates independent state, step consumes one item, and finish converts final state to the public result. An optional combine merges partial states. done enables source short-circuiting, and native carries planner metadata without changing Python execution.

Built-in collectors

Shared and approximate statistics (2.5.0)

Collectors.statistics("value", ddof=1) returns count, mean, variance, and std in one dictionary. It selects and converts each nonmissing value once. Selected None values are skipped; an undefined mean, variance, or standard deviation is None. This explicit collector shares its statistical state without changing the callback behavior of existing collectors. It has no state combiner.

These collectors are not included in PyPI 2.4.1. From a source checkout, install the sketch extra to use the optional Apache DataSketches C++ kernels:

from fpstreams import Collectors, flow

result = flow([1, 2, 3, None, 4]).collect(
    distribution=Collectors.approx_quantiles([0, 0.5, 1], k=200),
    cardinality=Collectors.approx_count_distinct(lg_k=12),
)

approx_quantiles preserves rank order in its result tuple. It skips None and converts real numeric inputs to finite double-precision values; conversion can round large integers. Empty input returns one None per requested rank. Complex values, including NumPy complex scalars with a zero imaginary part, are rejected before conversion. Rank parameters use the same real-number check. The KLL parameter k ranges from 8 to 65535. Estimates are stochastic and have rank error, which does not bound error in the returned numeric value. The DataSketches KLL documentation describes its empirical error model. Do not infer a fixed byte limit from k.

approx_count_distinct returns a floating-point HLL estimate, including 0.0 for empty input. Its lg_k ranges from 7 to 21. It accepts exact builtin bool, int, finite float, str, and bytes values; None is skipped. Equal numeric values such as True, 1, and 1.0 share an identity. Text and bytes have different identities, including empty values. Other objects and nonfinite floats are rejected. The adapter encodes scalars with type framing and exact integer ratios, then applies a versioned BLAKE2b digest before native hashing. This adds work per value and does not provide general Python object equality semantics.

Both factories import the optional dependency only when called. Their states support compatible partial merges: the left state is updated and the right is preserved. Algorithm and precision must match, and states must be independent. State serialization is not a public persistence contract. Merge order can affect estimates, so byte equality and exact reducer laws are not promised. Existing grouping creates fresh state per group; many groups can still consume substantial memory. These sketches do not certify uniqueness or replace exact collectors.

DataSketches 5.2.0 publishes wheels for CPython 3.11 through 3.13, but no 3.14 or 3.15 wheel. Installing the sketch extra on these versions requires a source build and its toolchain. The base library has no DataSketches dependency.

Collector composition and completion (2.4.0)

head(count) and tail(count) accept integers and objects implementing __index__. Each factory normalizes the count once, so the same collector can be reused without consulting the count provider again. Negative integer counts raise ValueError; floats, NaN, infinity, and other non-index values raise TypeError at construction. A zero head still pulls no items, while a zero tail consumes the source and returns an empty list.

Completed first() and head(count) states remain stable when a grouping adapter continues offering input. For example, Collectors.grouping_by(key, Collectors.first()) retains the first item per key, and a downstream head(0) retains an empty list for each encountered group. grouping_by and partitioning_by still read the full source and call downstream steps for every item without consulting its done hook. These built-in steps keep their completed result unchanged. Custom downstream callbacks continue receiving every input item.

Collectors.teeing() provides a state combiner when both downstream collectors declare one at construction. It combines the left states, then the right states. Downstream finishers and the result merger run only when the combined state is finished. Missing combiners leave combine=None; a live combiner removed later raises ValueError when needed.

An available combiner does not establish associativity or authorize parallel execution. explain_reduction() reports combine_declared separately from reducer laws. User combiners can mutate their input states; mutations cannot be rolled back if a later callback fails.

For custom reducers, fpstreams.testing.check_reducer_laws() samples identity, associativity, and partition equivalence. It closes the input before replaying examples and initializes independent states for the two associativity trees. Identity is sampled on both sides using separate states. Each sampled partition stops stepping when done is true. This differs from run_partitioned_reducer, which consumes and steps every input. A passing partition sample therefore does not establish that this execution function can run the reducer successfully. associativity_checked is false for fewer than three inputs; partition_equivalence_checked is false without layouts. The corresponding boolean results remain true when no example was checked for backward compatibility, and are not evidence that the property holds. The new checked flags do not participate in ReducerLawReport equality, preserving the previous comparison contract; inspect the flags explicitly when coverage matters. Its equivalent callback compares raw states for associativity and finished outputs for the other checks, so a finisher that hides state differences can also hide an identity failure. Passing sampled examples does not prove the laws for every input or permit reordering arbitrary callbacks. For example, floating-point addition may pass one associativity sample but produce a different result under another partition layout due to rounding. Partition layouts also accept integer-index objects; checked_partitions records their normalized Python integer values.

Flow.mean() returns None on an empty flow and uses compensated floating-point summation. Collectors.averaging() returns 0.0 on empty input and uses ordinary addition followed by division; rounding and result types depend on the operands. In 2.5.0, its result annotation is object: nonempty Decimal and Fraction inputs preserve those division result types, complex inputs return complex, and custom division can return any type. This annotation replaces the inaccurate float declaration; the runtime arithmetic and empty 0.0 are unchanged. Existing typed callers must narrow that result or migrate to the explicit interface below. For example, [1e16, 1.0, -1e16] yields 1/3 from mean() but 0.0 from averaging() in the Python paths. These operations do not automatically skip None. A collector's merge order can also affect floating-point rounding.

Explicit averaging operations (2.5.0)

Collectors.averaging_with() requires a callable selector and explicit zero, add, divide, and empty callbacks. The selector determines the input and selected-value types; accumulator and result types can differ. No result is converted to float. zero() creates fresh accumulator state per collection, divide(total, count) runs only for nonempty input, and empty() supplies the empty result. Both finish callbacks must return the declared result type, which can include None or another explicit alternative.

from decimal import Decimal
from fpstreams import Collectors, flow


def value(item: Decimal) -> Decimal:
    return item


def add(total: Decimal, item: Decimal) -> Decimal:
    return total + item


def divide(total: Decimal, count: int) -> Decimal:
    return total / count


average = Collectors.averaging_with(
    value, zero=Decimal, add=add, divide=divide, empty=Decimal,
)
assert average([Decimal("1.1"), Decimal("2.2")]) == Decimal("1.65")
assert average([]) == Decimal(0)
assert flow([Decimal("2")]).collect(average) == Decimal("2")

For records, supply an annotated selector such as def value(row: dict[str, Decimal]) -> Decimal: return row["amount"]. String and integer selectors remain available on legacy averaging(); the explicit interface uses callables so its selected-value type is known. A custom division result can also have a different type from its accumulator:

class Number:
    def __init__(self, value: int = 0) -> None:
        self.value = value

    def __add__(self, other: "Number") -> "Number":
        return Number(self.value + other.value)

    def __truediv__(self, count: int) -> str:
        return f"{self.value}/{count}"


def number(item: Number) -> Number:
    return item


def add_number(total: Number, item: Number) -> Number:
    return total + item


def divide_number(total: Number, count: int) -> str:
    return total / count


text_average = Collectors.averaging_with(
    number, zero=Number, add=add_number, divide=divide_number, empty=str,
)
assert text_average([Number(3), Number(4)]) == "7/2"
assert text_average([]) == ""

The resulting Collector supports direct calls, Flow.collect, AsyncFlow.collect, and existing Collector composition. collecting_and_then preserves its result type; the existing grouping_by annotation still includes its default list alternative in the result values. Partial-state merging is disabled unless combine(total_left, total_right) is supplied; counts are then added separately. The caller owns the callbacks' arithmetic, errors, precision, and merge laws. This interface provides an explicit typed migration; it does not infer the legacy factory's arbitrary operator results.

In the Python summarizing() path, an empty summary reports zero for its extrema. A summary containing only NaN has count 1 and NaN sum, while its minimum and maximum remain positive and negative infinity. With [NaN, 2.0], its extrema are both 2.0, while Flow.min() and Flow.max() return NaN for that order. With [2.0, NaN], the summary extrema and both Flow methods return 2.0. These are implementation-specific examples, not a guarantee for every backend or numeric type. For the Python paths tested with [float('inf'), float('-inf')], Flow.mean() and Collectors.averaging() return NaN; Python's statistics.fmean() raises ValueError for the same input. Select the contract your application requires.

In a tested Arrow int64 row aggregation, an empty sum is 0, a None value raises TypeError, and [2**63 - 1, 1] sums to the Python integer 2**63. For the same Arrow arrays, pyarrow.compute.sum returns null for empty input, skips nulls, and wraps that integer sum to -2**63. A tested NumPy int64 group sum also returns 2**63, including on the numpy_direct route, while numpy.sum wraps the fixed-width values. These examples do not guarantee the same behavior for every source, dtype, aggregation, or backend.

For the tested float64 Rows minimum, [NaN, 1.0] returns NaN and [1.0, NaN] returns 1.0; equal signed zeros retain the first input's sign. In the same samples, numpy.min returns NaN for either NaN order and the last zero's sign. Both python and auto reported planned:python for those examples; they do not demonstrate a NumPy-native extrema kernel.

fpstreams.Collectors

Build reusable collectors for containers, grouping, adaptation, and summaries.

statistics `classmethod`

statistics(
    selector: Selector | None = None,
    *,
    ddof: SupportsIndex = 0,
) -> Collector[T, Any, dict[str, Any]]

Compute count, mean, variance, and standard deviation with one conversion per value.

Parameters:

Name	Type	Description	Default
`selector`	`Selector \| None`	Value selector; `None` selects the whole item. Selected `None` is skipped.	`None`
`ddof`	`SupportsIndex`	Nonnegative integer subtracted from count for variance and standard deviation.	`0`

Returns:

Type	Description
`Collector[T, Any, dict[str, Any]]`	A collector returning count, mean, variance, and std. Undefined statistics are `None`. Numeric conversion and nonfinite behavior follow the exact statistics terminals. This collector does not declare a partial-state combiner.

approx_quantiles `classmethod`

approx_quantiles(
    ranks: Iterable[float],
    selector: Selector | None = None,
    *,
    k: SupportsIndex = 200,
) -> Collector[T, Any, tuple[float | None, ...]]

Estimate quantiles using the optional DataSketches double-precision KLL kernel.

Parameters:

Name	Type	Description	Default
`ranks`	`Iterable[float]`	Normalized ranks in [0, 1], frozen in supplied order at construction.	required
`selector`	`Selector \| None`	Value selector; selected `None` is skipped. Values must be finite real numbers and are converted to float, which can round large integers.	`None`
`k`	`SupportsIndex`	Integer precision in [8, 65535]; larger values trade memory for rank accuracy.	`200`

Returns:

Type	Description
`Collector[T, Any, tuple[float \| None, ...]]`	A mergeable collector returning a tuple of estimates, or one `None` per rank for empty input. Error concerns rank, not value; results are stochastic. Merging mutates only the left state and requires matching precision.

Raises:

Type	Description
`ImportError`	If the optional `fpstreams[sketch]` dependency is unavailable.
`TypeError`	If counts or numeric inputs have unsupported types.
`ValueError`	If precision, ranks, or selected numbers are out of range or nonfinite.

approx_count_distinct `classmethod`

approx_count_distinct(
    selector: Selector | None = None,
    *,
    lg_k: SupportsIndex = 12,
) -> Collector[T, Any, float]

Estimate distinct finite builtin scalars with an optional DataSketches HLL union.

Parameters:

Name	Type	Description	Default
`selector`	`Selector \| None`	Value selector; selected `None` is skipped. Supported values are exact builtin bool, int, finite float, str, and bytes. Equal numeric values share an identity; text and bytes remain distinct. Custom objects are rejected.	`None`
`lg_k`	`SupportsIndex`	Integer precision in [7, 21], controlling the register count as `2**lg_k`.	`12`

Returns:

Type	Description
`Collector[T, Any, float]`	A mergeable collector returning a floating-point estimate; empty input returns `0.0`. The stable, versioned scalar encoding precedes native hashing. Compatible merges mutate only the left state. State persistence is not an API.

Raises:

Type	Description
`ImportError`	If the optional `fpstreams[sketch]` dependency is unavailable.
`TypeError`	If precision or a selected value has an unsupported type.
`ValueError`	If precision is out of range or a selected float is nonfinite.

to_list `classmethod`

to_list() -> Collector[T, list[T], list[T]]

Build an order-preserving, mergeable collector for all input items.

Returns:

Type	Description
`Collector[T, list[T], list[T]]`	A reducer whose finished list is also its accumulated state.

to_set `classmethod`

to_set() -> Collector[T, set[T], set[T]]

Build a collector that retains each distinct hashable input item.

Returns:

Type	Description
`Collector[T, set[T], set[T]]`	A collector with set-union support for partial states.

to_tuple `classmethod`

to_tuple() -> Collector[T, list[T], tuple[T, ...]]

Build an order-preserving collector that finishes list state as a tuple.

Returns:

Type	Description
`Collector[T, list[T], tuple[T, ...]]`	A mergeable reducer returning an immutable tuple.

joining `staticmethod`

joining(
    delimiter: str = "",
) -> Collector[Any, list[str], str]

Build an order-preserving collector that joins each item's str value.

Conversion occurs as items are stepped. Empty input finishes as an empty string.

Parameters:

Name	Type	Description	Default
`delimiter`	`str`	The string inserted between collected values.	`''`

Returns:

Type	Description
`Collector[Any, list[str], str]`	A mergeable reducer that joins its accumulated strings with `delimiter`.

grouping_by `classmethod`

grouping_by(
    classifier: Selector,
    downstream: Collector[T, Any, R]
    | Callable[[Iterable[T]], R]
    | None = None,
) -> Collector[T, dict[Any, Any], dict[Any, R | list[T]]]

Build a collector that maintains one downstream state per selected key.

Only encountered keys appear in the result, in first-encounter order. Each item is stepped into its group even if the downstream collector has an early-completion predicate. A callable downstream is invoked at finish time with a materialized list for each group; omitting it collects group members into lists.

Parameters:

Name	Type	Description	Default
`classifier`	`Selector`	Selector producing a hashable group key for each item.	required
`downstream`	`Collector[T, Any, R] \| Callable[[Iterable[T]], R] \| None`	Per-group collector or callable, defaulting to list collection.	`None`

Returns:

Type	Description
`Collector[T, dict[Any, Any], dict[Any, R \| list[T]]]`	A collector that finishes every encountered group's state into a dictionary.

partitioning_by `classmethod`

partitioning_by(
    predicate: Selector,
    downstream: Collector[T, Any, R]
    | Callable[[Iterable[T]], R]
    | None = None,
) -> Collector[T, dict[bool, Any], dict[bool, R | list[T]]]

Build a collector with independently reduced false and true partitions.

Both Boolean keys are always present, even when one or both partitions receive no items. Each source item is stepped into one partition without consulting the downstream collector's early-completion predicate. Omitting downstream collects partition members into lists.

Parameters:

Name	Type	Description	Default
`predicate`	`Selector`	Selector whose truth value chooses the partition.	required
`downstream`	`Collector[T, Any, R] \| Callable[[Iterable[T]], R] \| None`	Collector or callable applied independently to both partitions.	`None`

Returns:

Type	Description
`Collector[T, dict[bool, Any], dict[bool, R \| list[T]]]`	A collector returning `{False: false_result, True: true_result}`.

mapping `classmethod`

mapping(
    mapper: Selector,
    downstream: Collector[U, Any, R]
    | Callable[[Iterable[U]], R],
) -> Collector[T, Any, R]

Build a collector that selects a new value before each downstream step.

The adapter preserves downstream initialization, finishing, combining, and early completion, but does not carry downstream native metadata or reducer-law type.

Parameters:

Name	Type	Description	Default
`mapper`	`Selector`	Selector applied once to each consumed source item.	required
`downstream`	`Collector[U, Any, R] \| Callable[[Iterable[U]], R]`	Collector or callable receiving mapped values.	required

Returns:

Type	Description
`Collector[T, Any, R]`	A collector that maps each source item before returning the downstream result.

filtering `classmethod`

filtering(
    predicate: Selector,
    downstream: Collector[T, Any, R]
    | Callable[[Iterable[T]], R],
) -> Collector[T, Any, R]

Build a collector that steps downstream only for predicate matches.

Rejected items leave downstream state unchanged. Initialization, finishing, combining, and early completion are preserved; native metadata and reducer-law type are not.

Parameters:

Name	Type	Description	Default
`predicate`	`Selector`	Selector evaluated for truth against each item.	required
`downstream`	`Collector[T, Any, R] \| Callable[[Iterable[T]], R]`	Collector or callable receiving matching items.	required

Returns:

Type	Description
`Collector[T, Any, R]`	A collector that filters source items before returning the downstream result.

flat_mapping `classmethod`

flat_mapping(
    mapper: Selector,
    downstream: Collector[U, Any, R]
    | Callable[[Iterable[U]], R],
) -> Collector[T, Any, R]

Build a collector that steps downstream over each item's expanded iterable.

Expansion stops as soon as the downstream state reports completion. A nested iterator exposing close is closed after exhaustion, early completion, or error. Downstream combining and early completion are preserved, while native metadata and reducer-law type are not.

Parameters:

Name	Type	Description	Default
`mapper`	`Selector`	Selector returning an iterable for each source item.	required
`downstream`	`Collector[U, Any, R] \| Callable[[Iterable[U]], R]`	Collector or callable receiving nested items.	required

Returns:

Type	Description
`Collector[T, Any, R]`	A collector that flattens source items before returning the downstream result.

collecting_and_then `classmethod`

collecting_and_then(
    downstream: Collector[T, Any, R]
    | Callable[[Iterable[T]], R],
    finisher: Callable[[R], U],
) -> Collector[T, Any, U]

Build a collector that applies one more transformation after downstream finish.

The downstream state machine, combiner, and early-completion behavior are preserved. finisher is called exactly once with the downstream public result.

Parameters:

Name	Type	Description	Default
`downstream`	`Collector[T, Any, R] \| Callable[[Iterable[T]], R]`	Collector or callable that produces the intermediate result.	required
`finisher`	`Callable[[R], U]`	Callable converting that intermediate result to the final value.	required

Returns:

Type	Description
`Collector[T, Any, U]`	An adapted collector returning `finisher(downstream_result)`.

Raises:

Type	Description
`TypeError`	If `finisher` is not callable.

teeing `classmethod`

teeing(
    left: Collector[T, Any, R] | Callable[[Iterable[T]], R],
    right: Collector[T, Any, U]
    | Callable[[Iterable[T]], U],
    merger: Callable[[R, U], V],
) -> Collector[T, _TeeState, V]

Build a collector that shares one source between two downstream collectors.

Each input is offered only to downstream states that are not already complete. Source consumption stops when both are complete, then merger receives their finished results in left-to-right order. Callable downstreams buffer their respective items. If both downstream collectors declare a combiner at construction, partial states can be combined left-to-right before finishing. The adapter reads live combiners and carries neither reducer laws nor native metadata.

Parameters:

Name	Type	Description	Default
`left`	`Collector[T, Any, R] \| Callable[[Iterable[T]], R]`	First collector or iterable-consuming callable.	required
`right`	`Collector[T, Any, U] \| Callable[[Iterable[T]], U]`	Second collector or iterable-consuming callable.	required
`merger`	`Callable[[R, U], V]`	A callable that merges two downstream results.	required

Returns:

Type	Description
`Collector[T, _TeeState, V]`	A short-circuiting collector returning the merger result.

Raises:

Type	Description
`TypeError`	If either downstream or `merger` is not callable as required.

counting `classmethod`

counting() -> Collector[T, int, int]

Build a constant-state reducer that increments once per input item.

Returns:

Type	Description
`Collector[T, int, int]`	A commutative reducer returning zero for empty input.

frequencies `classmethod`

frequencies() -> Collector[T, dict[T, int], dict[T, int]]

Count hashable inputs in first-encounter order, retaining one entry per value.

mode `classmethod`

mode() -> Collector[T, Any, T | None]

Collect the most frequent hashable value, breaking ties by first appearance.

Empty input returns None; None is otherwise counted like any other value. State retains one entry per distinct value and has no partial-state combiner.

summing `classmethod`

summing(
    selector: Selector | None = None,
) -> Collector[T, Any, Any]

Build a collector that adds selected values from a zero identity.

Parameters:

Name	Type	Description	Default
`selector`	`Selector \| None`	Value selector; `None` adds each whole input item.	`None`

Returns:

Type	Description
`Collector[T, Any, Any]`	A collector returning zero for empty input and supporting partial-state addition.

averaging `classmethod`

averaging(
    selector: Selector | None = None,
) -> Collector[T, tuple[Any, int], object]

Build a collector that tracks selected-value sum and count for a mean.

Empty input returns 0.0. Nonempty input divides the accumulated sum by count during finishing, and partial (sum, count) states can be combined component-wise. The result type is object because operands can define arbitrary division results. Use averaging_with to specify typed operations and an explicit empty result.

Parameters:

Name	Type	Description	Default
`selector`	`Selector \| None`	Value selector; `None` averages each whole input item.	`None`

Returns:

Type	Description
`Collector[T, tuple[Any, int], object]`	A collector returning the arithmetic mean or `0.0` for no values.

averaging_with `classmethod`

averaging_with(
    selector: Callable[[T], V],
    *,
    zero: Callable[[], S],
    add: Callable[[S, V], S],
    divide: Callable[[S, int], R],
    empty: Callable[[], R],
    combine: Callable[[S, S], S] | None = None,
) -> Collector[T, tuple[S, int], R]

Build a mean collector with explicit accumulator and result contracts.

Parameters:

Name	Type	Description	Default
`selector`	`Callable[[T], V]`	Typed value selector, including identity for whole input items.	required
`zero`	`Callable[[], S]`	Create an independent initial accumulator for each collection.	required
`add`	`Callable[[S, V], S]`	Add a selected value to the accumulator.	required
`divide`	`Callable[[S, int], R]`	Produce the nonempty result from the accumulator and count.	required
`empty`	`Callable[[], R]`	Produce the result for empty input, without calling `divide`.	required
`combine`	`Callable[[S, S], S] \| None`	Optional accumulator merger for partial states.	`None`

Returns:

Type	Description
`Collector[T, tuple[S, int], R]`	A collector preserving the supplied result type without numeric conversion. Its combiner is absent unless `combine` is supplied. Callers own the arithmetic and merge laws of their callbacks.

summarizing `classmethod`

summarizing(
    selector: Selector | None = None,
) -> Collector[T, SummaryStatistics, SummaryStatistics]

Build a collector returning mutable count, sum, extrema, and average state.

Selected values update SummaryStatistics directly. Empty input changes the minimum and maximum from infinities to 0.0; the sum, count, and derived average are already zero. The returned object is the final mutable state.

Parameters:

Name	Type	Description	Default
`selector`	`Selector \| None`	Value selector; `None` summarizes each whole input item.	`None`

Returns:

Type	Description
`Collector[T, SummaryStatistics, SummaryStatistics]`	A one-pass collector returning SummaryStatistics.

first `classmethod`

first() -> Collector[T, Any, T | None]

Build a collector that stops after and returns the first input item.

Empty input returns None. A first item whose value is itself None is still treated as a completed result. Later items offered by an adapter leave that result unchanged.

Returns:

Type	Description
`Collector[T, Any, T \| None]`	A collector that consumes at most one item.

last `classmethod`

last() -> Collector[T, Any, T | None]

Build a collector that consumes all input and returns its last item.

Empty input returns None.

Returns:

Type	Description
`Collector[T, Any, T \| None]`	A collector retaining only the most recently consumed item.

head `classmethod`

head(
    count: SupportsIndex,
) -> Collector[T, list[T], list[T]]

Build a collector that stops after retaining the first count items.

A zero count completes before pulling from the source.

Parameters:

Name	Type	Description	Default
`count`	`SupportsIndex`	Non-negative integer maximum, including objects implementing `__index__`.	required

Returns:

Type	Description
`Collector[T, list[T], list[T]]`	An early-stopping collector returning an encounter-ordered list.

Raises:

Type	Description
`TypeError`	If `count` does not provide an integer index.
`ValueError`	If `count` is negative.

tail `classmethod`

tail(
    count: SupportsIndex,
) -> Collector[T, deque[T], list[T]]

Build a bounded-state collector for the final count input items.

The source is fully consumed. Once the deque is full, each new item discards the oldest; a zero count retains nothing.

Parameters:

Name	Type	Description	Default
`count`	`SupportsIndex`	Non-negative integer maximum, including objects implementing `__index__`.	required

Returns:

Type	Description
`Collector[T, deque[T], list[T]]`	A collector finishing its bounded deque as an encounter-ordered list.

Raises:

Type	Description
`TypeError`	If `count` does not provide an integer index.
`ValueError`	If `count` is negative.

only `classmethod`

only() -> Collector[T, list[T], T | None]

Build a collector that enforces zero-or-one input cardinality.

Empty input returns None, one item is returned directly, and a second item completes collection early so finishing can raise without pulling a third.

Returns:

Type	Description
`Collector[T, list[T], T \| None]`	A collector retaining no more than two items.

Raises:

Type	Description
`ValueError`	If the input contains more than one item.

to_dict `classmethod`

to_dict(
    key: Selector,
    value: Selector,
    *,
    on_duplicate: Literal[
        "error", "first", "last"
    ] = "error",
) -> Collector[T, dict[K, V], dict[K, V]]

Build a dictionary collector with an explicit duplicate-key policy.

On duplicates, "error" raises before selecting the duplicate value, "first" preserves the existing entry without selecting another value, and "last" replaces it. Selector errors and unhashable keys propagate.

Parameters:

Name	Type	Description	Default
`key`	`Selector`	Selector deriving each dictionary key.	required
`value`	`Selector`	Selector deriving each dictionary value.	required
`on_duplicate`	`Literal['error', 'first', 'last']`	One of `"error"`, `"first"`, or `"last"`.	`'error'`

Returns:

Type	Description
`Collector[T, dict[K, V], dict[K, V]]`	A collector preserving key insertion order in its result dictionary.

Raises:

Type	Description
`ValueError`	If `on_duplicate` is not a supported policy.
`DuplicateKeyError`	If a repeated key is encountered under `"error"`.

to_columns `staticmethod`

to_columns() -> Collector[
    Mapping[str, Any], _ColumnsState, dict[str, list[Any]]
]

Build a collector that transposes variably shaped mapping rows into columns.

Columns retain field encounter order. Fields introduced by later rows are backfilled with None, and fields absent from later rows append None, so every column has one entry per input row.

Returns:

Type	Description
`Collector[Mapping[str, Any], _ColumnsState, dict[str, list[Any]]]`	A collector returning a dictionary of equally sized column lists.

Aggregators

agg creates named, single-pass aggregations:

Call	Result
`agg.count()`	Number of input items
`agg.count_where(predicate)`	Number of matching items
`agg.any(predicate)` / `agg.all(predicate)`	Boolean checks
`agg.sum(selector)` / `agg.mean(selector)`	Sum or arithmetic mean
`agg.variance(selector, ddof=1)`	Variance
`agg.std(selector, ddof=1)`	Standard deviation
`agg.count_distinct(selector)`	Number of distinct values
`agg.min(selector)` / `agg.max(selector)`	Smallest or largest value
`agg.min_by(selector)` / `agg.max_by(selector)`	Whole item with the smallest or largest key (2.4.0)
`agg.top(count, key=...)` / `agg.bottom(count, key=...)`	Up to `count` whole items in descending or ascending key order (2.4.0)
`agg.first(selector)` / `agg.last(selector)`	Boundary values
`agg.collect(selector, into=list)`	Values collected with a constructor

Omit selector to aggregate the items themselves. A string selector reads a record field; expression and callable selectors are also accepted.

In the 2.5.0 Python path, agg.count_distinct checks equality across hashable and unhashable values. It retains distinct values, may call equality across categories, and can take O(H × U) cross-category comparisons for H hashable and U unhashable values, with O(H + U) retained references. The Python path compares previously accepted values to the incoming value using list membership semantics. A hash TypeError takes the unhashable path; other hash errors and comparison errors propagate.

min_by and max_by require a selector and return the original winning item. They select a key once per item, keep the first item on ties, and return None for empty input. Each state retains one key and one item. Keys use Python's < or > comparison, including its errors and NaN behavior. Apply a key normalization yourself when missing or incomparable values need a policy.

These factories work with Flow.aggregate, AsyncFlow.aggregate, and global, grouped, sorted, or spilled row aggregation. Spill restores serialized records under its existing contract. The factories do not declare merge laws for arbitrary Python keys. See the latest-record recipe.

top and bottom return lists, retain at most count items per state, and keep input order for equal keys. The key runs once per item; payloads are never compared. Count must be a non-negative integer. Empty input and zero count return []; zero count skips key calls and does not pull an independent input. Grouped aggregation still reads the source to discover groups. Use consistently ordered keys: Python heap and sorting comparisons determine the result and propagate comparison errors. These aggregators declare no merge laws. See the per-group ranking recipe.

fpstreams.Aggregator `dataclass`

Bases: Collector[Any, Any, Any]

A collector accepted by named aggregate terminals and native fusion.

Collecting and aggregation

Collector type inference (2.5.0)

Collector

fpstreams.Collector dataclass

Built-in collectors

Shared and approximate statistics (2.5.0)

Collector composition and completion (2.4.0)

Explicit averaging operations (2.5.0)

fpstreams.Collectors

statistics classmethod

approx_quantiles classmethod

approx_count_distinct classmethod

to_list classmethod

to_set classmethod

to_tuple classmethod

joining staticmethod

grouping_by classmethod

partitioning_by classmethod

mapping classmethod

filtering classmethod

flat_mapping classmethod

collecting_and_then classmethod

teeing classmethod

counting classmethod

frequencies classmethod

mode classmethod

summing classmethod

averaging classmethod

averaging_with classmethod

summarizing classmethod

first classmethod

last classmethod

head classmethod

tail classmethod

only classmethod

to_dict classmethod

to_columns staticmethod

Aggregators

fpstreams.Aggregator dataclass

fpstreams.Collector `dataclass`

statistics `classmethod`

approx_quantiles `classmethod`

approx_count_distinct `classmethod`

to_list `classmethod`

to_set `classmethod`

to_tuple `classmethod`

joining `staticmethod`

grouping_by `classmethod`

partitioning_by `classmethod`

mapping `classmethod`

filtering `classmethod`

flat_mapping `classmethod`

collecting_and_then `classmethod`

teeing `classmethod`

counting `classmethod`

frequencies `classmethod`

mode `classmethod`

summing `classmethod`

averaging `classmethod`

averaging_with `classmethod`

summarizing `classmethod`

first `classmethod`

last `classmethod`

head `classmethod`

tail `classmethod`

only `classmethod`

to_dict `classmethod`

to_columns `staticmethod`

fpstreams.Aggregator `dataclass`