[Python] Fix DataFrame skew() and kurtosis() with a single non-empty partition

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

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',

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