damccorm commented on code in PR #41140:
URL: https://github.com/apache/beam/pull/41140#discussion_r4525606788
##########
sdks/python/apache_beam/dataframe/frames.py:
##########
@@ -1854,8 +1854,11 @@ def _corr_aligned(self, other, min_periods):
std_x = self.std()
std_y = other.std()
cov = self._cov_aligned(other, min_periods)
+ # Like pandas, a series with no variance has a NaN correlation.
return cov.apply(
- lambda cov, std_x, std_y: cov / (std_x * std_y), args=[std_x, std_y])
+ lambda cov, std_x, std_y: cov / (std_x * std_y)
+ if std_x * std_y else float('nan'),
+ args=[std_x, std_y])
Review Comment:
Checking exact non-zero (`if std_x * std_y`) and `if args[ix] > 0` (in
`DataFrame.corr` below) breaks when a constant column contains a non-dyadic
float like `0.1` instead of `1.0` (e.g., `df = pd.DataFrame({'a': np.arange(10,
dtype=float), 'b': 0.1})`).
When `combine_moments` (in `Series.var`) and `combine_co_moments` (in
`_cov_aligned`) combine partitions of different sizes (such as `n=3` and
`n=2`), `delta = s / n - datum.s / datum.n` evaluates `(0.1 * 3) / 3 - (0.1 *
2) / 2 = 1.39e-17 != 0`, producing a tiny positive variance (`~2.5e-35 > 0`)
and `std` (`~7.6e-18 != 0`). Because `_corr_aligned` also computes
`self.std()`, `other.std()`, and `self._cov_aligned(other, min_periods)` as
three separate aggregations with different partitioning requirements
(`Arbitrary()` for `std` vs `Index()` for `cov`), changing `'b': 1.0` to `'b':
0.1` in `test_corr_no_variance` causes all three checks to fail:
- `df.corr()` sets the `(b, b)` diagonal to `1.0` instead of `NaN`
(`args[ix] > 0` is `True`).
- `df.a.corr(df.b)` returns `0.173` instead of `NaN`.
- `df.b.corr(df.b)` returns `1.98` (`> 1.0`) instead of `NaN`.
##########
sdks/python/apache_beam/dataframe/frames.py:
##########
@@ -3123,8 +3133,8 @@ def cov(self, min_periods, ddof):
arg_indices = []
for col in columns:
arg_indices.append((col, col))
- std = self[col].std(ddof)
- args.append(std.apply(lambda x: x*x, 'square'))
+ x = self[col].dropna()
+ args.append(x._cov_aligned(x, min_periods, ddof))
Review Comment:
Two issues with switching the diagonal of `DataFrame.cov` (and
`DataFrame.corr` above) to `x = self[col].dropna(); x._cov_aligned(x, ...)`:
1. **Unnecessary `Index()` shuffle per column:** Originally, `DataFrame.cov`
used `self[col].std(ddof)` on the diagonal because single-column variance
(`DeferredSeries.var`) only requires `partitionings.Arbitrary()` (no index
shuffle). Calling `x._cov_aligned(x, ...)` passes `[x._expr, x._expr]` into a
`ComputedExpression` with `requires_partition_by=partitionings.Index()`, which
forces the stage planner (`transforms.py`) to insert a `CoGroupByKey`
index-repartitioning stage for each column just to compute single-column
variance. Adding a `min_periods` parameter (or internal `_var(ddof=1,
min_periods=None)` helper) to `DeferredSeries.var` with
`requires_partition_by=partitionings.Arbitrary()` would avoid the shuffle.
2. **`ddof` edge cases in `_cov_aligned` (`n <= ddof` and `ddof=0, n=1`):**
In `_cov_aligned` (line 1897), the guard is `if n < max(2, ddof, min_periods or
0): return float('nan')`:
- When `ddof >= 2` and `n == ddof` (e.g., a 2-row DataFrame with
`df.cov(ddof=2)`), `n < ddof` is `False` and `c / (n - ddof)` raises
`ZeroDivisionError: float division by zero` (whereas `Series.var` checks `if n
<= ddof:`).
- When `ddof == 0` and `n == 1` (e.g., a 1-row DataFrame with
`df.cov(ddof=0)`), hardcoding `2` in `max(2, ...)` returns `NaN` instead of
`0.0` (whereas pandas and `Series.var(ddof=0)` return `0.0`).
--
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]
