[Data] pyarrow.lib.ArrowInvalid when aggregating over Python list of dicts

[cadedaniel]

Ray 2.2 on MacOS, pyarrow==6.0.1, python 3.7

stacktrace:

  File "python/ray/_raylet.pyx", line 830, in ray._raylet.execute_task
  File "python/ray/_raylet.pyx", line 834, in ray._raylet.execute_task
  File "/Users/cade/miniconda3/envs/anyscale/lib/python3.7/site-packages/ray/data/grouped_dataset.py", line 58, in map
    parts = [BlockAccessor.for_block(p).combine(key, aggs) for p in partitions]
  File "/Users/cade/miniconda3/envs/anyscale/lib/python3.7/site-packages/ray/data/grouped_dataset.py", line 58, in <listcomp>
    parts = [BlockAccessor.for_block(p).combine(key, aggs) for p in partitions]
  File "/Users/cade/miniconda3/envs/anyscale/lib/python3.7/site-packages/ray/data/_internal/arrow_block.py", line 515, in combine
    return builder.build()
  File "/Users/cade/miniconda3/envs/anyscale/lib/python3.7/site-packages/ray/data/_internal/table_block.py", line 98, in build
    tables = [self._table_from_pydict(self._columns)]
  File "/Users/cade/miniconda3/envs/anyscale/lib/python3.7/site-packages/ray/data/_internal/arrow_block.py", line 123, in _table_from_pydict
    return pyarrow.Table.from_pydict(columns)
  File "pyarrow/table.pxi", line 1724, in pyarrow.lib.Table.from_pydict
  File "pyarrow/table.pxi", line 2368, in pyarrow.lib._from_pydict
  File "pyarrow/array.pxi", line 341, in pyarrow.lib.asarray
  File "pyarrow/array.pxi", line 315, in pyarrow.lib.array
  File "pyarrow/array.pxi", line 39, in pyarrow.lib._sequence_to_array
  File "pyarrow/error.pxi", line 143, in pyarrow.lib.pyarrow_internal_check_status
  File "pyarrow/error.pxi", line 99, in pyarrow.lib.check_status
pyarrow.lib.ArrowInvalid: Could not convert {'name': 'C', 'amount': 10, 'country': 'C1'} with type ArrowRow: did not recognize Python value type when inferring an Arrow data type

repro script:

#!/usr/bin/env python3

import ray
from ray.data.aggregate import AggregateFn

data = [
    {'name': 'A', 'amount': 100, 'country': 'C1'},
    {'name': 'B', 'amount': 200, 'country': 'C2'},
    {'name': 'C', 'amount': 10, 'country': 'C1'},
    {'name': 'D', 'amount': 500, 'country': 'C2'},
    {'name': 'E', 'amount': 400, 'country': 'C3'},
]

ds = ray.data.from_items(data)
ds = ds.groupby('country')

result = ds.aggregate(AggregateFn(
    init=lambda k: [],
    accumulate_row=lambda a, r: a + [r],
    merge=lambda a1, a2: a1['amount'] + a2['amount'],
    finalize=lambda a: a
))

full log https://gist.github.com/cadedaniel/1080563aae30309aef98505aef9fc6bc
pip freeze https://gist.github.com/cadedaniel/480a95d8d29da7795ebd19f092253b44

[daikeshi]

Hey, I also hit a similar error when I use a third-party library (TDigest) with Ray Data's AggregateFn to compute percentiles.

My environment:
Python: 3.8.16
Ray: 2.3.0
PyArrow: 7.0.0
OS: MacOS 13.2.1

Here's the repo script

from typing import Dict, Iterator, Optional, Union

from ray.data.aggregate import _AggregateOnKeyBase
from ray.data.block import Block, BlockAccessor, KeyFn
from ray.data.row import TableRow
from tdigest import TDigest

RowType = Union[int, float, TableRow]

def _initialize_tdigest(k: Optional[float]) -> TDigest:
    digest = TDigest()
    if k is not None:
        digest.update(k)
    return digest

### Aggregator Class
class PercentileAggregator(_AggregateOnKeyBase):
    def __init__(self, on: Optional[KeyFn] = None):
        self._set_key_fn(on)

        def init(row: RowType) -> TDigest:
            if isinstance(row, TableRow):
                return _initialize_tdigest(row.get(on, None))
            elif isinstance(row, int) or isinstance(row, float) or row is None:
                return _initialize_tdigest(row)
            else:
                raise TypeError("Not a supported data row type: {} ({})".format(row, type(row)))

        def merge(left: TDigest, right: TDigest) -> TDigest:
            sum = left + right
            return sum

        def accumulate_block(aggregator: TDigest, block: Block) -> TDigest:
            rows: Iterator[RowType] = BlockAccessor.for_block(block).iter_rows()
            for row in rows:
                if isinstance(row, TableRow):
                    v = row.get(on, None)
                    if v is not None:
                        aggregator.update(v)
                elif isinstance(row, int) or isinstance(row, float) or row is None:
                    if row is not None:
                        aggregator.update(row)
                else:
                    raise TypeError("Not a supported data row type: {} ({})".format(row, type(row))) 

            aggregator.compress() 
            return aggregator

        super().__init__(
            init=init,
            accumulate_block=accumulate_block,
            merge=merge,
            name=f"t-digest({str(on)})",
        )

### Code to use the custom aggregator
ray.init(num_cpus=2)
df = pd.DataFrame({"a": [1, 2, 3, 4, 5], "b": [6, 7, 8, 9, 10]})
table = pa.Table.from_pandas(df)
data = ray.data.from_arrow(table)
aggregator_a = PercentileAggregator("a")
aggregator_b = PercentileAggregator("b")
digests = data.aggregate(aggregator_a, aggregator_b)

assert digest_a.n == 5
assert digest_b.n == 5

When I ran the above script, I'm hitting the following error:

(map pid=6450) 2023-03-11 00:36:58,536  INFO worker.py:772 -- Task failed with retryable exception: TaskID(18b2ad3c688fb947ffffffffffffffffffffffff01000000).
(map pid=6450) Traceback (most recent call last):
(map pid=6450)   File "python/ray/_raylet.pyx", line 857, in ray._raylet.execute_task
(map pid=6450)   File "python/ray/_raylet.pyx", line 861, in ray._raylet.execute_task
(map pid=6450)   File "/Users/keshi/miniforge3/envs/hendrix/lib/python3.8/site-packages/ray/data/grouped_dataset.py", line 58, in map
(map pid=6450)     parts = [BlockAccessor.for_block(p).combine(key, aggs) for p in partitions]
(map pid=6450)   File "/Users/keshi/miniforge3/envs/hendrix/lib/python3.8/site-packages/ray/data/grouped_dataset.py", line 58, in <listcomp>
(map pid=6450)     parts = [BlockAccessor.for_block(p).combine(key, aggs) for p in partitions]
(map pid=6450)   File "/Users/keshi/miniforge3/envs/hendrix/lib/python3.8/site-packages/ray/data/_internal/arrow_block.py", line 527, in combine
(map pid=6450)     return builder.build()
(map pid=6450)   File "/Users/keshi/miniforge3/envs/hendrix/lib/python3.8/site-packages/ray/data/_internal/table_block.py", line 96, in build
(map pid=6450)     tables = [self._table_from_pydict(self._columns)]
(map pid=6450)   File "/Users/keshi/miniforge3/envs/hendrix/lib/python3.8/site-packages/ray/data/_internal/arrow_block.py", line 123, in _table_from_pydict
(map pid=6450)     return pyarrow.Table.from_pydict(columns)
(map pid=6450)   File "pyarrow/table.pxi", line 1868, in pyarrow.lib.Table.from_pydict
(map pid=6450)   File "pyarrow/table.pxi", line 2658, in pyarrow.lib._from_pydict
(map pid=6450)   File "pyarrow/array.pxi", line 342, in pyarrow.lib.asarray
(map pid=6450)   File "pyarrow/array.pxi", line 316, in pyarrow.lib.array
(map pid=6450)   File "pyarrow/array.pxi", line 39, in pyarrow.lib._sequence_to_array
(map pid=6450)   File "pyarrow/error.pxi", line 143, in pyarrow.lib.pyarrow_internal_check_status
(map pid=6450)   File "pyarrow/error.pxi", line 99, in pyarrow.lib.check_status
(map pid=6450) pyarrow.lib.ArrowInvalid: Could not convert <T-Digest: n=5, centroids=5> with type TDigest: did not recognize Python value type when inferring an Arrow data type

[stale]

Hi, I'm a bot from the Ray team :)

To help human contributors to focus on more relevant issues, I will automatically add the stale label to issues that have had no activity for more than 4 months.

If there is no further activity in the 14 days, the issue will be closed!

• If you'd like to keep the issue open, just leave any comment, and the stale label will be removed!
• If you'd like to get more attention to the issue, please tag one of Ray's contributors.

You can always ask for help on our discussion forum or Ray's public slack channel.

[stale]

Hi again! The issue will be closed because there has been no more activity in the 14 days since the last message.

Please feel free to reopen or open a new issue if you'd still like it to be addressed.

Again, you can always ask for help on our discussion forum or Ray's public slack channel.

Thanks again for opening the issue!

[wpm]

I am seeing the same thing which I posted about in the discussion forum.

I think this is a recent regression. I'm suddenly seeing it with a few of my data structures that have been working for months.

This seems to have started when I recently upgraded from Ray 2.9.1 to Ray 2.22.0.