Dataset in Pandas Returns Arrow Argument When Materializing

How severe does this issue affect your experience of using Ray?

  • Medium: It contributes to significant difficulty to complete my task, but I can work around it.

I have Dataset object created by actors that are using the pandas batch format. When I materialize it, I get an error about Ray failing to infer an Arrow type.

The code in which I am hitting this is proprietary and somewhat complex, so I can’t post it here. I haven’t created a minial minimal repro scenario yet, but it appears to be caused by a groupby/aggregate operation.

from operator import attrgetter

from import Dataset
from import AggregateFn

def collate(dataset: Dataset) -> Dataset:
    dataset = dataset.add_column("document_name", lambda f: f["document"].apply(attrgetter("name")))
    dataset = dataset.groupby("document_name").aggregate(
            init=lambda column: MyObject(),
            accumulate_row=lambda a, r: a | r["my_object"],
            merge=lambda a1, a2: a1 | a2,
    return dataset.drop_columns(["document_name"])

The dataset coming in to this function has two columns: document and my_object. I want to group the my_object items by document. Ray won’t let me use the document objects temselves as group key, so I temporarily add a document_name column that contains a unique string returned by the name property of document.

MyObject defines an or operator that combines two MyObject instances into a single one.

When I run either take() or to_pandas() on a dataset that has had this aggregation, I get the following error.

  File "<ipython-input-1-ac5b7c801a02>", line 1, in <module>
  File "/Users/bill.mcneill/anaconda3/envs/my-project/lib/python3.10/site-packages/ray/data/", line 2408, in take
    for row in limited_ds.iter_rows():
  File "/Users/bill.mcneill/anaconda3/envs/my-project/lib/python3.10/site-packages/ray/data/", line 245, in _wrapped_iterator
    for batch in batch_iterable:
  File "/Users/bill.mcneill/anaconda3/envs/my-project/lib/python3.10/site-packages/ray/data/", line 162, in _create_iterator
    block_iterator, stats, blocks_owned_by_consumer = self._to_block_iterator()
  File "/Users/bill.mcneill/anaconda3/envs/my-project/lib/python3.10/site-packages/ray/data/_internal/iterator/", line 33, in _to_block_iterator
    block_iterator, stats, executor = ds._plan.execute_to_iterator()
  File "/Users/bill.mcneill/anaconda3/envs/my-project/lib/python3.10/site-packages/ray/data/", line 86, in handle_trace
    raise e.with_traceback(None) from SystemException()
ray.exceptions.RayTaskError(ArrowInvalid): ray::reduce() (pid=59917, ip=
  File "/Users/bill.mcneill/anaconda3/envs/my-project/lib/python3.10/site-packages/ray/data/_internal/planner/exchange/", line 70, in reduce
    return BlockAccessor.for_block(mapper_outputs[0]).aggregate_combined_blocks(
  File "/Users/bill.mcneill/anaconda3/envs/my-project/lib/python3.10/site-packages/ray/data/_internal/", line 606, in aggregate_combined_blocks
    blocks = TableBlockAccessor.normalize_block_types(blocks, "arrow")
  File "/Users/bill.mcneill/anaconda3/envs/my-project/lib/python3.10/site-packages/ray/data/_internal/", line 296, in normalize_block_types
    results = [BlockAccessor.for_block(block).to_arrow() for block in blocks]
  File "/Users/bill.mcneill/anaconda3/envs/my-project/lib/python3.10/site-packages/ray/data/_internal/", line 296, in <listcomp>
    results = [BlockAccessor.for_block(block).to_arrow() for block in blocks]
  File "/Users/bill.mcneill/anaconda3/envs/my-project/lib/python3.10/site-packages/ray/data/_internal/", line 266, in to_arrow
    return pyarrow.table(self._table)
  File "pyarrow/table.pxi", line 5147, in pyarrow.lib.table
  File "/Users/bill.mcneill/anaconda3/envs/my-project/lib/python3.10/site-packages/pandas/core/", line 1025, in __arrow_c_stream__
    table = pa.Table.from_pandas(self, schema=requested_schema)
  File "pyarrow/table.pxi", line 3869, in pyarrow.lib.Table.from_pandas
  File "/Users/bill.mcneill/anaconda3/envs/my-project/lib/python3.10/site-packages/pyarrow/", line 613, in dataframe_to_arrays
    arrays = [convert_column(c, f)
  File "/Users/bill.mcneill/anaconda3/envs/my-project/lib/python3.10/site-packages/pyarrow/", line 613, in <listcomp>
    arrays = [convert_column(c, f)
  File "/Users/bill.mcneill/anaconda3/envs/my-project/lib/python3.10/site-packages/pyarrow/", line 600, in convert_column
    raise e
  File "/Users/bill.mcneill/anaconda3/envs/my-project/lib/python3.10/site-packages/pyarrow/", line 594, in convert_column
    result = pa.array(col, type=type_, from_pandas=True, safe=safe)
  File "pyarrow/array.pxi", line 340, in pyarrow.lib.array
  File "pyarrow/array.pxi", line 86, in pyarrow.lib._ndarray_to_array
  File "pyarrow/error.pxi", line 91, in pyarrow.lib.check_status
pyarrow.lib.ArrowInvalid: ('Could not convert MyObject with type MyObject: did not recognize Python value type when inferring an Arrow data type', 'Conversion failed for column my_object with type object')

It seems like Ray is running into trouble because it is confusing a pandas batch format with an arrow one. This is somewhere down in the Ray data code when it is building iterators over Blocks. I’ve stepped through but haven’t been able to figure out what is going on.

Am I making an obvious mistake? Can you give me any pointers on debugging this? I’ll try to create a minimal repro scenario, but I’ve already burned a week on this an am getting pressure to move on.