[Feature Request]: [Python] Add Variance, StandardDeviation, Covariance, and LinearRegression to apache_beam.transforms.combiners

Ouverte
  • Quelque chose ne va pas ?
#41,171 0 commentaires 0 réactions 0 personnes assignées Voir sur GitHub

Les mainteneurs répondent en général sous 1 jour

Personne n'a encore pris cette issue.

Évaluation

Difficulté
5/5
Temps estimé
Plus d'une semaine
Accessibilité débutants
45/100
Type d'issue
Fonctionnalité
Clarté
Plutôt claire
Activité
Active
Stack technique
python
Domaine
data

Piste de recherche

Start by reading the Python SDK's apache_beam.transforms.combiners implementation and its existing combiner tests. The issue specifies four statistical combiners, stable distributed accumulator merging, multiple input shapes, and a structured regression result, so scope the work and tests around those requirements. Done means the new combiners handle the requested inputs and produce the stated statistics across distributed merges.

Rédigé par le modèle d'indexation à partir du texte de l'issue.

Description

P2 python
What would you like to happen?

Currently, apache_beam.transforms.combiners provides standard aggregation transforms such as Count, Mean, Min, Max, Top, and Sample. However, common statistical and bivariate regression aggregations are missing from the Python SDK:

  • Variance (sample and population)
  • StandardDeviation (sample and population)
  • Covariance (sample and population for (x, y) pairs or structured records)
  • LinearRegression (ordinary least squares slope, intercept, r, r2, and means)

While users can compute sample means with combiners.Mean.Globally(), calculating variance, standard deviation, covariance, or linear regression currently requires writing custom multi-pass accumulators or pulling in NumPy/SciPy inside custom CombineFns, which can be tricky to make numerically stable across distributed worker combines.

We propose adding first-class, numerically stable combiners in apache_beam.transforms.combiners:

import apache_beam as beam
from apache_beam.transforms import combiners

# 1. Variance & Standard Deviation
with beam.Pipeline() as p:
  numbers = p | beam.Create([10.0, 20.0, 30.0, 40.0, 50.0])
  sample_var = numbers | 'SampleVar' >> combiners.Variance.Globally()
  pop_stddev = numbers | 'PopStdDev' >> combiners.StandardDeviation.Globally(sample=False)

# 2. Covariance
with beam.Pipeline() as p:
  points = p | beam.Create([(1.0, 2.0), (2.0, 4.0), (3.0, 5.0), (4.0, 7.0)])
  cov = points | combiners.Covariance.Globally()

# 3. Ordinary Least Squares Linear Regression
with beam.Pipeline() as p:
  ad_data = p | beam.Create([
      {'spend': 10, 'revenue': 25.0},
      {'spend': 20, 'revenue': 41.0},
      {'spend': 30, 'revenue': 62.0},
  ])
  model = ad_data | combiners.LinearRegression.Globally(x='spend', y='revenue')
  # Emits RegressionResult(count=3, slope=1.85, intercept=5.67, r=0.998, r2=0.997, mean_x=20.0, mean_y=42.67)
Key Technical Details
  1. Numerical Stability: Uses single-pass, numerically stable Welford updating equations for both variance and bivariate cross-products, and Chan et al. pairwise merging for distributed accumulator merges (merge_accumulators).
  2. Flexible Input Shapes:
    • Variance and StandardDeviation accept numeric types (int, float, np.number).
    • Covariance and LinearRegression accept 2-tuples/sequences (x, y) or key-value/dict mappings with x='col_x', y='col_y'.
  3. Structured Output: LinearRegression returns a typed NamedTuple (RegressionResult) containing (count, slope, intercept, r, r2, mean_x, mean_y).
  4. Runner & Streaming Parity: Fully pure Python, runner-agnostic, zero extra dependencies, and works out-of-the-box in batch, streaming, DirectRunner, Prism, and Dataflow.
Issue Priority

Priority: 2 (default / most feature requests should be filed as P2)

Issue Components
  • Component: Python SDK
  • Component: beam-core
Langage dominant
Java
Étoiles
8.7k
Forks
4.8k
Merge moyen
2 j 12 h
PR mergées (30 j)
371

Préparer son environnement

Par où commencer

  1. Lisez l'issue en entier, puis le guide de contribution du projet.
  2. Signalez en commentaire que vous la prenez — cela évite que deux personnes fassent le même travail.
  3. Forkez le dépôt et travaillez sur une branche.
  4. Ouvrez une pull request qui référence le numéro de l'issue.

Autres issues de apache/beam

Toutes les issues de apache/beam

Issues similaires

Plus d'issues Java

Recevez les nouvelles issues par e-mail

Un résumé court des issues GitHub adaptées aux débutants.