[Datasets] GroupedDataset.map_groups() Doesn't Support Callable Classes

[thatcort]

What happened + What you expected to happen

The GroupedDataset method map_groups() is missing support for callable classes.

Attempting to use a callable class with map_groups() results in the following error:

Map Progress (1 actors 1 pending):  88%|████████▊ | 7/8 [00:36<00:04,  4.69s/it]2022-06-30 20:43:20,980	ERROR worker.py:95 -- Unhandled error (suppress with 'RAY_IGNORE_UNHANDLED_ERRORS=1'): ray::BlockWorker.map_block_nosplit() (pid=73870, ip=172.28.0.2, repr=<ray.data.impl.compute.BlockWorker object at 0x7ffb4136d5d0>)
  File "/usr/local/lib/python3.7/dist-packages/ray/data/impl/compute.py", line 185, in map_block_nosplit
    return _map_block_nosplit(block, fn, input_files)
  File "/usr/local/lib/python3.7/dist-packages/ray/data/impl/compute.py", line 341, in _map_block_nosplit
    for new_block in fn(block):
  File "/usr/local/lib/python3.7/dist-packages/ray/data/dataset.py", line 355, in transform
    applied = fn(view)
  File "/usr/local/lib/python3.7/dist-packages/ray/data/grouped_dataset.py", line 259, in group_fn
    applied = fn(group)
TypeError: __init__() takes 1 positional argument but 2 were given

https://ray-distributed.slack.com/archives/C01DLHZHRBJ/p1656626491260859?thread_ts=1656616666.457479&cid=C01DLHZHRBJ

Versions / Dependencies

This is using Ray 1.13.0 and Pyarrow 6.0.1 on Python 3.7.13

Reproduction script

class SentenceSplitter:
  def __init__(self):
    self.nlp = spacy.load("en_core_web_sm")
  
  def __call__(self, batch):
    doc = self.nlp(batch['text'][0])
    docIds = [batch['docId'][0] for i in range(len(sents))]
    return pa.Table.from_pydict({'docId': docIds, 'text': list(doc.sents)})

ds = ray.data.read_parquet(fileName)
ds = ds.groupby('docId').map_groups(SentenceSplitter, compute='actors', batch_format="pyarrow")

Issue Severity

No response

[jianoaix]

For reference, map_batches(): https://sourcegraph.com/github.com/ray-project/ray@master/-/blob/python/ray/data/dataset.py?L314