diff --git a/CHANGES.md b/CHANGES.md index 61b1547e025f..0e49f51683fe 100644 --- a/CHANGES.md +++ b/CHANGES.md @@ -150,6 +150,7 @@ * (Java/Python) The SQL `SqlTransformSchemaTransformProvider` configuration schema (`schematransform:org.apache.beam:sql_transform:v1`) now uses only portable types so it can be discovered from Python: `dialect` is a string, `tableproviders` is an array of strings, and the unused `parameters` field was removed. Invalid dialects and unknown table providers are rejected with a clear error ([#34613](https://github.com/apache/beam/issues/34613)). * (Python) An unknown pipeline option is no longer given the value of a later argument that does not directly follow it, which made `Pipeline.run()` and `get_all_options(retain_unknown_options=True)` fail with `expected one argument` for flags like `--unknown_flag --job_name=foo bar` ([#22434](https://github.com/apache/beam/issues/22434)). * (Python) Fixed `WriteToJson` silently rounding floats to 10 decimal places: it now defaults pandas' `double_precision` to 15, so values such as `0.1234567890123` and `1e-11` are written correctly. Floats that need 16-17 significant digits may still be rounded ([#41102](https://github.com/apache/beam/issues/41102)). +* (Python) Fixed the DataFrame API `skew()` and `kurtosis()` failing with `UnboundLocalError` when all of the data is in one partition ([#41283](https://github.com/apache/beam/issues/41283)). * Fixed X (Java/Python) ([#X](https://github.com/apache/beam/issues/X)). ## Security Fixes diff --git a/sdks/python/apache_beam/dataframe/frames.py b/sdks/python/apache_beam/dataframe/frames.py index e6387af10219..2dcc95e9333c 100644 --- a/sdks/python/apache_beam/dataframe/frames.py +++ b/sdks/python/apache_beam/dataframe/frames.py @@ -1753,9 +1753,7 @@ def combine_moments(data): elif m2 == 0: return float(0) else: - return combined_n * math.sqrt(combined_n - 1) / (combined_n - - 2) * m3 / ( - m2**(3 / 2)) + return n * math.sqrt(n - 1) / (n - 2) * m3 / (m2**(3 / 2)) moments = expressions.ComputedExpression( 'compute_moments', @@ -1827,12 +1825,9 @@ def combine_moments(data): elif m2 == 0: return float(0) else: - return (((combined_n + 1) * (combined_n) * (combined_n - 1)) / - ((combined_n - 2) * - (combined_n - 3))) * (m4 / - (m2)**2) - ((3 * (combined_n - 1)**2) / - ((combined_n - 2) * - (combined_n - 3))) + return (((n + 1) * (n) * (n - 1)) / + ((n - 2) * (n - 3))) * (m4 / (m2)**2) - ((3 * (n - 1)**2) / + ((n - 2) * (n - 3))) moments = expressions.ComputedExpression( 'compute_moments', diff --git a/sdks/python/apache_beam/dataframe/frames_test.py b/sdks/python/apache_beam/dataframe/frames_test.py index 84ea8be1f2cc..b324bf72095b 100644 --- a/sdks/python/apache_beam/dataframe/frames_test.py +++ b/sdks/python/apache_beam/dataframe/frames_test.py @@ -926,6 +926,13 @@ def test_series_cov_corr(self): self._run_test(lambda s: s.kurtosis(), s) self._run_test(lambda s: s.kurt(), s) + def test_series_skew_kurtosis_single_partition(self): + # With distributed=False all the data is in one partition, so the moments + # are never merged. + s = pd.Series([x**3 for x in range(1, 20)]) + self._run_test(lambda s: s.skew(), s, distributed=False) + self._run_test(lambda s: s.kurtosis(), s, distributed=False) + def test_dataframe_cov_corr(self): df = pd.DataFrame(np.random.randn(20, 3), columns=['a', 'b', 'c']) df.loc[df.index[:5], 'a'] = np.nan