dhanusharer opened a new issue, #40814:
URL: https://github.com/apache/beam/issues/40814

   ### What happened?
   
   ### Description
   
   Calling `duplicated()` on any `DeferredSeries` in the Beam DataFrame API 
(e.g. `s.duplicated(keep=False)` or `s.duplicated(keep="any")`) completely 
fails and raises a `WontImplementError`.
   
   ### Root Cause
   
   In 
[`apache_beam/dataframe/frames.py:1980-1991`](https://github.com/apache/beam/blob/master/sdks/python/apache_beam/dataframe/frames.py#L1980-L1991):
   
   ```python
     @frame_base.with_docs_from(pd.Series)
     @frame_base.args_to_kwargs(pd.Series)
     @frame_base.populate_defaults(pd.Series)
     @frame_base.maybe_inplace
     def duplicated(self, keep):
       # Re-use the DataFrame based duplcated, extract the series back out
       df = self._wrap_in_df()
   
       return df.duplicated(keep=keep)[df.columns[0]]
   
   ### Steps to Reproduce
   import pandas as pd
   import apache_beam as beam
   from apache_beam.dataframe.convert import to_dataframe
   
   with beam.Pipeline() as p:
       pcoll = p | beam.Create([{'a': 1}, {'a': 2}, {'a': 1}])
       df = to_dataframe(pcoll)
       # Crashes with WontImplementError:
       result = df['a'].duplicated(keep=False)
   ### Traceback
   apache_beam.dataframe.frame_base.WontImplementError: Indexing a series with 
key of type <class 'str'> is not supported because it produces a non-deferred 
result.
   For more information see https://s.apache.org/dataframe-non-deferred-result.
   
   ### Proposed Fix
   -    # Re-use the DataFrame based duplcated, extract the series back out
   -    df = self._wrap_in_df()
   -
   -    return df.duplicated(keep=keep)[df.columns[0]]
   +    # Re-use the DataFrame based duplicated
   +    df = self._wrap_in_df()
   +    result = df.duplicated(keep=keep)
   +    if self.name is not None:
   +      return result.rename(self.name)
   +    return result
   
   
   ### Issue Priority
   
   Priority: 2 (default / most bugs should be filed as P2)
   
   ### Issue Components
   
   - [x] Component: Python SDK
   - [ ] Component: Java SDK
   - [ ] Component: Go SDK
   - [ ] Component: Typescript SDK
   - [ ] Component: IO connector
   - [ ] Component: Beam YAML
   - [ ] Component: Beam examples
   - [ ] Component: Beam playground
   - [ ] Component: Beam katas
   - [ ] Component: Website
   - [ ] Component: Infrastructure
   - [ ] Component: Spark Runner
   - [ ] Component: Flink Runner
   - [ ] Component: Prism Runner
   - [ ] Component: Twister2 Runner
   - [ ] Component: Hazelcast Jet Runner
   - [ ] Component: Google Cloud Dataflow Runner


-- 
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]

Reply via email to