[python] - update deferredSeries.duplicated logic

sdks/python/apache_beam/dataframe/frames.py

@@ -1966,10 +1966,12 @@ def duplicated(self, keep):
     a Beam-specific option that guarantees only one duplicate will be kept, but
     unlike ``"first"`` and ``"last"`` it makes no guarantees about _which_
     duplicate element is kept."""
-    # Re-use the DataFrame based duplcated, extract the series back out
+    # Re-use the DataFrame based duplicated
     df = self._wrap_in_df()
-
-    return df.duplicated(keep=keep)[df.columns[0]]
+    result = df.duplicated(keep=keep)
+    if self.name is not None:
+      return result.rename(self.name)
+    return result
 
   @frame_base.with_docs_from(pd.Series)
   @frame_base.args_to_kwargs(pd.Series)
@@ -1983,8 +1985,10 @@ def drop_duplicates(self, keep):
     duplicate element is kept."""
     # Re-use the DataFrame based drop_duplicates, extract the series back out
     df = self._wrap_in_df()
-
-    return df.drop_duplicates(keep=keep)[df.columns[0]]
+    result = df.drop_duplicates(keep=keep)[df.columns[0]]
+    if self.name is None:
+      return result.rename(None)
+    return result
 
   @frame_base.with_docs_from(pd.Series)
   @frame_base.args_to_kwargs(pd.Series)

sdks/python/apache_beam/dataframe/frames_test.py

@@ -1211,9 +1211,14 @@ def test_duplicates_with_nan(self):
 
     s = pd.Series([1.0, 1.0, np.nan, 2.0], name='a')
     self._run_test(lambda s: s.drop_duplicates(keep=False), s)
+    self._run_test(lambda s: s.duplicated(keep=False), s)
     self._run_test(lambda s: s.nunique(), s)
     self._run_test(lambda s: s.nunique(dropna=False), s)
 
+    s_unnamed = pd.Series([1.0, 1.0, np.nan, 2.0])
+    self._run_test(lambda s: s.drop_duplicates(keep=False), s_unnamed)
+    self._run_test(lambda s: s.duplicated(keep=False), s_unnamed)
+
   def test_nunique_categorical_with_unused_categories(self):
     s = pd.Series(['a', 'b', 'a'], dtype=pd.CategoricalDtype(['a', 'b', 'c']))
     self._run_test(lambda s: s.nunique(), s)
@@ -2633,6 +2638,17 @@ def test_duplicated_keep_any(self):
     # Verify that the result is the same as conventional duplicated
     self.assert_frame_data_equivalent(result, df.duplicated())
 
+  def test_series_duplicated_keep_any(self):
+    df = pd.DataFrame({
+        'brand': ['Yum Yum', 'Yum Yum', 'Indomie', 'Indomie', 'Indomie'],
+        'style': ['cup', 'cup', 'cup', 'pack', 'pack'],
+        'rating': [4, 4, 3.5, 15, 5]
+    })
+
+    result = self._evaluate(lambda df: df.brand.duplicated(keep='any'), df)
+
+    self.assert_frame_data_equivalent(result, df.brand.duplicated())
+
   def test_get_dummies_not_categoricaldtype(self):
     # Should not work because series is not a CategoricalDtype type
     with self.assertRaisesRegex(

sdks/python/apache_beam/dataframe/pandas_doctests_test.py

@@ -533,6 +533,11 @@ def test_series_tests(self):
                 's.drop_duplicates()',
                 "s.drop_duplicates(keep='last')",
             ],
+            'pandas.core.series.Series.duplicated': [
+                'animals.duplicated()',
+                "animals.duplicated(keep='first')",
+                "animals.duplicated(keep='last')",
+            ],
             'pandas.core.series.Series.reindex': ['*'],
             'pandas.core.series.Series.autocorr': ['*'],
             'pandas.core.series.Series.repeat': ['s.repeat([1, 2, 3])'],
@@ -576,7 +581,6 @@ def test_series_tests(self):
             # Skipped idxmax/idxmin due an issue with the test framework
             'pandas.core.series.Series.idxmin': ['s.idxmin()'],
             'pandas.core.series.Series.idxmax': ['s.idxmax()'],
-            'pandas.core.series.Series.duplicated': ['*'],
             # Relies on setting index.
             'pandas.core.series.Series.rename_axis': ['*'],
             'pandas.core.series.Series.set_axis': ['*'],