Spark

Finding a threshold over a Spark DataFrame. See Running on Spark for the guide.

Importing this module does not import PySpark, so it is safe to reference from code that may run without the [spark] extra installed. PySpark is imported when a SparkThresher is constructed, and its absence is reported as a BackendDependencyError.

The entry point

thresher.spark.SparkThresher

SparkThresher(
    algorithm_name: str = "hist",
    algorithm_params: dict[str, Any] | None = None,
    labels: tuple[Any, Any] | None = None,
    verbose: bool = False,
    verbosity: str | None = None,
)

Find the optimal threshold over a Spark DataFrame.

Mirrors Thresher, but takes a DataFrame and the names of two columns instead of two sequences, and never collects the rows.

Example

from thresher.spark import SparkThresher SparkThresher().optimize_threshold(df, "probability", "label") # doctest: +SKIP 0.4306640625

Configure the search.

Parameters:

Name	Type	Description	Default
algorithm_name	str	'hist' (the default) or 'exact', or any of their synonyms. hist returns a summary bounded by its bin count, so it is the one that stays cheap however large the data is.	'hist'
algorithm_params	dict[str, Any] | None	passed through to the algorithm. hist reads no_of_bins (default 1024); exact reads none. A key the chosen algorithm does not read raises ConfigurationError rather than being ignored.	None
labels	tuple[Any, Any] | None	your two class labels, negative first, if they are not -1 and 1 - for example (0, 1).	None
verbose	bool	log what is being aggregated. verbose=True means verbosity='debug', the same as it does on Thresher.	False
verbosity	str | None	how much this instance reports - 'debug', 'info', 'warning' (the default), 'error' or 'critical'. Applies for the duration of each optimize_threshold call.	None

Note

No progress bar is offered, deliberately. The counting happens on executors, where there is nobody watching a terminal, and the driver's share of the work is a sweep over a few thousand bins - see thresher.progress.

Raises:

Type	Description
UnknownAlgorithmError	if the name matches no algorithm at all.
ConfigurationError	if it names an algorithm that cannot run as an aggregation, or if verbosity names no known level. Both are ValueError.
BackendDependencyError	if PySpark is not installed.

Source code in src/thresher/spark.py

def __init__(
    self,
    algorithm_name: str = "hist",
    algorithm_params: dict[str, Any] | None = None,
    labels: tuple[Any, Any] | None = None,
    verbose: bool = False,
    verbosity: str | None = None,
) -> None:
    """Configure the search.

    Args:
        algorithm_name: `'hist'` (the default) or `'exact'`, or any of their synonyms.
            `hist` returns a summary bounded by its bin count, so it is the one that
            stays cheap however large the data is.
        algorithm_params: passed through to the algorithm. `hist` reads `no_of_bins`
            (default 1024); `exact` reads none. A key the chosen algorithm does not
            read raises `ConfigurationError` rather than being ignored.
        labels: your two class labels, negative first, if they are not -1 and 1 -
            for example `(0, 1)`.
        verbose: log what is being aggregated. `verbose=True` means
            `verbosity='debug'`, the same as it does on `Thresher`.
        verbosity: how much this instance reports - `'debug'`, `'info'`, `'warning'`
            (the default), `'error'` or `'critical'`. Applies for the duration of each
            `optimize_threshold` call.

    Note:
        No progress bar is offered, deliberately. The counting happens on executors,
        where there is nobody watching a terminal, and the driver's share of the work
        is a sweep over a few thousand bins - see `thresher.progress`.

    Raises:
        UnknownAlgorithmError: if the name matches no algorithm at all.
        ConfigurationError: if it names an algorithm that cannot run as an
            aggregation, or if `verbosity` names no known level. Both are `ValueError`.
        BackendDependencyError: if PySpark is not installed.
    """
    _require_pyspark()

    resolved = algorithm.retrieve_by_alias(algorithm_name)
    if resolved.id not in SUPPORTED_ALGORITHMS:
        raise ConfigurationError(
            NOT_DISTRIBUTABLE.format(name=algorithm_name, available=", ".join(SUPPORTED_ALGORITHMS))
        )

    self.algorithm = resolved
    self.algorithm_params = algorithm_params or {}
    # The same check the in-memory interface runs: a key this algorithm does not read
    # would otherwise leave the default in place across an entire cluster run.
    validate_algorithm_params(resolved, self.algorithm_params)
    self.labels = labels
    self.verbosity = log.resolve_verbosity(verbosity, verbose)

optimize_threshold

optimize_threshold(
    df: DataFrame,
    score_col: str = "score",
    label_col: str = "label",
) -> float

Find the threshold that classifies the most rows correctly.

Parameters:

Name	Type	Description	Default
df	DataFrame	the DataFrame holding the scores and their true classes. It is read once or twice depending on the algorithm, and never collected.	required
score_col	str	name of the column holding the scores.	'score'
label_col	str	name of the column holding the ground-truth classes.	'label'

Returns:

Type	Description
float	The threshold, computed identically to what the in-memory algorithm would
float	return for the same data.

Raises:

Type	Description
EmptyInputError	if the DataFrame has no rows.
UndefinedScoresError	if any score is null or NaN.
MissingLabelsError	if any label is null.
UnexpectedLabelsError	if any label is neither of the two declared classes.
SingleClassError	if only one class is present, leaving nothing to separate.
InsufficientDataError	if the aggregation came back empty.

Note

Those refusals are the same ones the in-memory path makes, deliberately. Until 0.7.1 none of them existed here: a row whose label matched neither class - a null, a third value, a typo - was counted as a negative simply because it was not a positive, and a null or NaN score was quietly filed in the top bin. Both returned a plausible threshold computed from data the in-memory path would have refused outright.

Source code in src/thresher/spark.py

def optimize_threshold(
    self, df: "DataFrame", score_col: str = "score", label_col: str = "label"
) -> float:
    """Find the threshold that classifies the most rows correctly.

    Args:
        df: the DataFrame holding the scores and their true classes. It is read once
            or twice depending on the algorithm, and never collected.
        score_col: name of the column holding the scores.
        label_col: name of the column holding the ground-truth classes.

    Returns:
        The threshold, computed identically to what the in-memory algorithm would
        return for the same data.

    Raises:
        EmptyInputError: if the DataFrame has no rows.
        UndefinedScoresError: if any score is null or NaN.
        MissingLabelsError: if any label is null.
        UnexpectedLabelsError: if any label is neither of the two declared classes.
        SingleClassError: if only one class is present, leaving nothing to separate.
        InsufficientDataError: if the aggregation came back empty.

    Note:
        Those refusals are the same ones the in-memory path makes, deliberately. Until
        0.7.1 none of them existed here: a row whose label matched neither class - a
        null, a third value, a typo - was counted as a negative simply because it was
        not a positive, and a null or NaN score was quietly filed in the top bin. Both
        returned a plausible threshold computed from data the in-memory path would have
        refused outright.
    """
    with log.verbosity(self.verbosity):
        return self._optimize_threshold(df, score_col, label_col)

The module

thresher.spark

Find a threshold over a Spark DataFrame, without bringing the data to the driver.

The other entry points take a sequence in memory. That is the wrong shape for data living in HDFS or S3: collecting a billion rows to one machine to sort them defeats the point of having a cluster.

This does the counting where the data already is. Spark performs one aggregation - a groupBy and a pair of sums - and returns a summary bounded by the resolution rather than by the row count. The driver then runs exactly the same sweep the in-memory algorithms use, over that summary.

a billion rows  ->  [Spark: group and count]  ->  ~1,024 rows  ->  [driver: sweep]

Two algorithms fit that shape:

hist Groups by bin index, so the summary is no_of_bins rows however large the input. Nothing else here is bounded that way, which makes it the one to reach for.

exact Groups by distinct score, so the summary is one row per distinct score. Exact, and fine when scores are rounded probabilities; a warning is logged when the distinct count is large enough for the collect to be the expensive part.

The rest are not offered, for the same reason the Ray backend does not distribute them: a backend may change where the work happens, never the answer. ls is O(n²) in candidates, and sgrid, gen and sgd each draw their own random subsamples, so sharding would change which samples are read and therefore the result.

PySpark is an optional dependency: pip install 'thresher-py[spark]'. It also needs a JVM, which pip cannot install for you.

The sweeps it reuses

The deciding half of each supported algorithm is a plain function over counts, which is what makes running it on a summary from the cluster identical to running it in memory.

thresher.algs.histogram.compute.sweep_bins

sweep_bins(
    negatives: Sequence[int],
    positives: Sequence[int],
    lowest: float,
    highest: float,
    progress_bar: bool = False,
) -> tuple[float, int]

Find the best threshold from binned class counts.

The deciding half of the algorithm, kept separate from the counting half. It needs only the per-bin totals and the range they cover, which is a summary whose size is the bin count and nothing else - so whatever produced those counts, a single pass here or an aggregation across a cluster, this function makes the decision and the answers agree.

Parameters:

Name	Type	Description	Default
negatives	Sequence[int]	count of negative samples in each bin, lowest bin first.	required
positives	Sequence[int]	count of positive samples in each bin, in the same order.	required
lowest	float	the smallest score the bins cover.	required
highest	float	the largest. The two are taken rather than a span because lowest + span does not always reconstruct it: for scores rounded to a few decimal places - the ordinary case - 0.065 + (0.997 - 0.065) is 0.9969999999999999, and a threshold there classifies the largest samples positive while the counting below has them negative. span is derived from the pair, once.	required
progress_bar	bool	draw a progress bar on stderr while sweeping.	False

Returns:

Type	Description
float	The best threshold expressible on a bin edge, and how many samples it classifies
int	correctly - a count the threshold is guaranteed to achieve.

Raises:

Type	Description
InsufficientDataError	if there are no bins to sweep.

Source code in src/thresher/algs/histogram/compute.py

def sweep_bins(
    negatives: Sequence[int],
    positives: Sequence[int],
    lowest: float,
    highest: float,
    progress_bar: bool = False,
) -> tuple[float, int]:
    """Find the best threshold from binned class counts.

    The deciding half of the algorithm, kept separate from the counting half. It needs only
    the per-bin totals and the range they cover, which is a summary whose size is the bin
    count and nothing else - so whatever produced those counts, a single pass here or an
    aggregation across a cluster, this function makes the decision and the answers agree.

    Args:
        negatives: count of negative samples in each bin, lowest bin first.
        positives: count of positive samples in each bin, in the same order.
        lowest: the smallest score the bins cover.
        highest: the largest. The two are taken rather than a span because `lowest + span`
            does not always reconstruct it: for scores rounded to a few decimal places -
            the ordinary case - `0.065 + (0.997 - 0.065)` is `0.9969999999999999`, and a
            threshold there classifies the largest samples positive while the counting
            below has them negative. `span` is derived from the pair, once.
        progress_bar: draw a progress bar on stderr while sweeping.

    Returns:
        The best threshold expressible on a bin edge, and how many samples it classifies
        correctly - a count the threshold is guaranteed to achieve.

    Raises:
        InsufficientDataError: if there are no bins to sweep.
    """
    bins = len(negatives)
    if bins == 0 or bins != len(positives):
        raise InsufficientDataError("Bin counts are missing or do not line up.")

    span = highest - lowest

    total_positive = sum(positives)

    # Everything classified positive: the only split that needs a threshold below the data.
    # `None` stands for it, since it sits below every bin rather than between two.
    best_correct = total_positive
    best_index: int | None = None

    negatives_behind = 0
    positives_behind = 0

    with make_progress(bins, "Sweeping bins", enabled=progress_bar) as bar:
        for index in range(bins):
            negatives_behind += negatives[index]
            positives_behind += positives[index]

            bar.update(index + 1)

            # A threshold at this bin's upper edge: everything up to here is predicted
            # negative, everything above it positive.
            correct = negatives_behind + (total_positive - positives_behind)

            if correct > best_correct:
                best_correct = correct
                best_index = index

    # Resolved once, for the winner only - the search below is far too expensive to run
    # for every candidate, and only the chosen split needs a threshold at all.
    if best_index is None:
        best_threshold = math.nextafter(lowest, -math.inf)
    elif best_index == bins - 1:
        # The topmost edge is the largest score itself: nothing exceeds it, which is how
        # "classify everything as negative" is expressed. Taken as given rather than
        # rebuilt from the span - see the note on `highest` above.
        best_threshold = highest
    else:
        best_threshold = _boundary_threshold(lowest, span, best_index + 1, bins)

    return best_threshold, best_correct

thresher.algs.exact.compute.sweep_class_counts

sweep_class_counts(
    counts: Mapping[float, tuple[int, int]],
    progress_bar: bool = False,
) -> tuple[float, int]

Find the best threshold from class counts, without seeing the samples.

The whole of the exact search lives here. It takes only "how many of each class sit at each distinct score", which is a summary bounded by the number of distinct scores rather than by the number of rows - so whatever produced those counts, in this process or across a cluster, the decision is made by this one function and the answers agree.

Parameters:

Name	Type	Description	Default
counts	Mapping[float, tuple[int, int]]	distinct score mapped to its (negatives, positives).	required
progress_bar	bool	draw a progress bar on stderr while sweeping.	False

Returns:

Type	Description
tuple[float, int]	The best threshold and how many samples it classifies correctly.

Raises:

Type	Description
InsufficientDataError	if there are no counts to sweep.

Source code in src/thresher/algs/exact/compute.py

def sweep_class_counts(
    counts: Mapping[float, tuple[int, int]], progress_bar: bool = False
) -> tuple[float, int]:
    """Find the best threshold from class counts, without seeing the samples.

    The whole of the exact search lives here. It takes only "how many of each class sit at
    each distinct score", which is a summary bounded by the number of distinct scores
    rather than by the number of rows - so whatever produced those counts, in this process
    or across a cluster, the decision is made by this one function and the answers agree.

    Args:
        counts: distinct score mapped to its `(negatives, positives)`.
        progress_bar: draw a progress bar on stderr while sweeping.

    Returns:
        The best threshold and how many samples it classifies correctly.

    Raises:
        InsufficientDataError: if there are no counts to sweep.
    """
    if not counts:
        raise InsufficientDataError("At least one score is needed to evaluate a threshold.")

    ordered = sorted(counts)
    distinct = len(ordered)
    total_positive = sum(positives for _, positives in counts.values())

    negatives_behind = 0
    positives_behind = 0
    best_correct = -1
    best_threshold = float(ordered[-1])

    with make_progress(distinct, "Sweeping scores", enabled=progress_bar) as bar:
        for index, score in enumerate(ordered):
            negatives_here, positives_here = counts[score]
            negatives_behind += negatives_here
            positives_behind += positives_here

            bar.update(index + 1)

            # Everything up to and including this score is predicted negative, everything
            # above it positive. Runs of equal scores are indivisible, which is automatic
            # here: they are one entry.
            correct = negatives_behind + (total_positive - positives_behind)

            if correct > best_correct:
                best_correct = correct
                # Below the largest score, sit between the two scores being separated; at
                # the largest score, sit on it, which predicts everything negative.
                best_threshold = (score + ordered[index + 1]) / 2 if index + 1 < distinct else float(score)

    # The one split no threshold inside the data can express: everything classified
    # positive, which needs a threshold strictly below the smallest score. nextafter gives
    # the largest float that qualifies, so the answer stays as close to the data as the
    # representation allows. Considered last, and only taken on a strict improvement, so a
    # threshold outside the input range is never returned merely to break a tie.
    if total_positive > best_correct:
        best_correct = total_positive
        best_threshold = math.nextafter(ordered[0], -math.inf)

    return best_threshold, best_correct