[Datasets] Add Pandas-native groupby and sorting. #26313
Merged
Conversation
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
clarkzinzow
requested review from
ericl,
scv119,
jjyao and
jianoaix
as code owners
clarkzinzow
changed the title
clarkzinzow
force-pushed
the
datasets/feat/pandas-groupby-sort-impl
branch
from
0e30293
to
f7d7d93
Compare
jjyao
approved these changes
Jul 6, 2022
| bounds = table[col].searchsorted(boundaries) | ||
| last_idx = 0 | ||
| for idx in bounds: | ||
| # Slices need to be copied to avoid including the base table |
There was a problem hiding this comment.
Is this comment relevant for pandas table as well?
There was a problem hiding this comment.
It's not, let me remove that!
| """ | ||
| if key is not None and not isinstance(key, str): | ||
| raise ValueError( | ||
| "key must be a string or None when aggregating on Arrow blocks, but " |
2 tasks
clarkzinzow
force-pushed
the
datasets/feat/pandas-groupby-sort-impl
branch
from
f7d7d93
to
b116873
Compare
c21
approved these changes
Jul 20, 2022
| ) | ||
|
|
||
| if self._table.shape[0] == 0: | ||
| # If the pyarrow table is empty we may not have schema |
There was a problem hiding this comment.
pyarrow table -> pandas DataFrame?
|
|
||
| def combine(self, key: KeyFn, aggs: Tuple[AggregateFn]) -> "pandas.DataFrame": | ||
| # TODO (kfstorm): A workaround to pass tests. Not efficient. | ||
| return BlockAccessor.for_block(self.to_arrow()).combine(key, aggs).to_pandas() | ||
| """Combine rows with the same key into an accumulator. |
There was a problem hiding this comment.
This looks like very similar to ArrowBlockAccessor.combine, except the builder is different. Shall we refactor them into a single method?
This is a non-blocking comment.
| stats = BlockExecStats.builder() | ||
| blocks = [b for b in blocks if b.shape[0] > 0] | ||
| if len(blocks) == 0: | ||
| ret = PandasBlockAccessor._empty_table() |
There was a problem hiding this comment.
why not just self._empty_table() as above?
|
|
||
| @staticmethod | ||
| def merge_sorted_blocks( | ||
| blocks: List["pandas.DataFrame"], key: "SortKeyT", _descending: bool | ||
| blocks: List[Block[T]], key: "SortKeyT", _descending: bool | ||
| ) -> Tuple["pandas.DataFrame", BlockMetadata]: |
There was a problem hiding this comment.
what's blocking us to change return type to Tuple[Block[T], BlockMetadata] same as ArrowBlockAccessor?
| ) | ||
| next_row = None | ||
| builder = PandasBlockBuilder() | ||
| while True: |
There was a problem hiding this comment.
nit: this merge-aggregate loop is also mostly same as Arrow's function. Non-blocking comment as well.
|
Discussed offline, feel free to merge it as it is now, and I can add a followup PR to address the comments. |
Rohan138
pushed a commit
to Rohan138/ray
that referenced
this pull request
Jul 28, 2022
This PR adds a Pandas-native implementation of groupby and sorting for Pandas blocks. Before this PR, we were converting to Arrow, doing groupbys + aggregations and sorting in Arrow land, and then converting back to Pandas; this to-from-Arrow conversion was happening both on the map side and the reduce side, which was very inefficient for Pandas blocks (many extra table copies). By adding Pandas-native groupby + sorting, we should see a decrease in memory utilization and faster performance when using the AIR preprocessors. Signed-off-by: Rohan138 <rapotdar@purdue.edu>
Stefan-1313
pushed a commit
to Stefan-1313/ray_mod
that referenced
this pull request
Aug 18, 2022
This PR adds a Pandas-native implementation of groupby and sorting for Pandas blocks. Before this PR, we were converting to Arrow, doing groupbys + aggregations and sorting in Arrow land, and then converting back to Pandas; this to-from-Arrow conversion was happening both on the map side and the reduce side, which was very inefficient for Pandas blocks (many extra table copies). By adding Pandas-native groupby + sorting, we should see a decrease in memory utilization and faster performance when using the AIR preprocessors. Signed-off-by: Stefan van der Kleij <s.vanderkleij@viroteq.com>
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
This PR adds a Pandas-native implementation of groupby and sorting for Pandas blocks. Before this PR, we were converting to Arrow, doing groupbys + aggregations and sorting in Arrow land, and then converting back to Pandas; this to-from-Arrow conversion was happening both on the map side and the reduce side, which was very inefficient for Pandas blocks (many extra table copies). By adding Pandas-native groupby + sorting, we should see a decrease in memory utilization and faster performance when using the AIR preprocessors.
Related issue number
Closes #21296
Checks
scripts/format.shto lint the changes in this PR.