jk-shah opened a new issue, #41171:
URL: https://github.com/apache/beam/issues/41171
### 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 `CombineFn`s, 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`:
```python
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
- [x] Component: Python SDK
- [x] Component: beam-core
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
