PyArrow Error when processing records with missing columns with flat_map

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

  • High: It blocks me to complete my task.

Hey folks I’m running into an issue when applying a flat_map on a ray dataset that contains s3 paths. During the __call__ op, I am downloading the data from the s3 path (which contains a bunch of jsonl records) and processing jsonl records materialized as dictionaries where each record can have a slightly different columns and then I return a nested list of dictionaries. Say these are the records I’m processing in the flatmap.

record1 -> {"a": 1, "b": 2}
record2 -> {"a": 1, "b": 2, "c": 3}
record3 -> {"a": 1, "b": 2}

I am returning:

return [{"items": [record1, record2, record3]}]

to get around the fact that I might have slightly different schema of records. So instead of returning a list(record1, record2, ...)
I’m returning list({“items”: list(record1, record2, …)}) which I’m hoping pyarrow Table can be stored with a single column “items”. I am however getting the following error when I do ds.take(5)

pyarrow.lib.ArrowTypeError: struct fields don't match or are in the wrong order

Here is the full stacktrace in the log:

ray.exceptions.RayTaskError(ArrowTypeError): ray::FlatMap(ShardProcessor)() (pid=857231, ip=, actor_id=e7d1e719651963e3e38b8bc506000000, repr=MapWorker(FlatMap(ShardProcessor)))
    yield from _map_task(
  File "/lib/python3.10/site-packages/ray/data/_internal/execution/operators/", line 427, in _map_task
    for b_out in map_transformer.apply_transform(iter(blocks), ctx):
  File "/lib/python3.10/site-packages/ray/data/_internal/execution/operators/", line 395, in __call__
  File "/lib/python3.10/site-packages/ray/data/_internal/", line 73, in next
    block_to_yield =
  File "/lib/python3.10/site-packages/ray/data/_internal/", line 64, in build
  File "/lib/python3.10/site-packages/ray/data/_internal/", line 133, in build
    return self._concat_tables(tables)
  File "/lib/python3.10/site-packages/ray/data/_internal/", line 148, in _concat_tables
    return transform_pyarrow.concat(tables)
  File "/lib/python3.10/site-packages/ray/data/_internal/arrow_ops/", line 263, in concat
    table = pyarrow.concat_tables(blocks, promote_options="default")
  File "pyarrow/table.pxi", line 5962, in pyarrow.lib.concat_tables
  File "pyarrow/error.pxi", line 154, in pyarrow.lib.pyarrow_internal_check_status
  File "pyarrow/error.pxi", line 91, in pyarrow.lib.check_status
pyarrow.lib.ArrowTypeError: struct fields don't match or are in the wrong order

Here is my processing function I use as part of a callable class in flat_map:

    def __call__(self, row: Dict[str, Any]) -> List[Dict[str, List[Dict[str, Any]]]]:
        import jsonlines
        import copy
        import gc

        data: List[Dict[str, List[Dict[str, Any]]]] = []
        chunks = []

        path = row['path']
        fileobj = self.download_fileobj(s3_url=path)
        with jsonlines.Reader(fileobj) as reader:
            for obj in reader:
                if len(chunks) >= self.chunk_size:
                    data.append({"items": chunks})
                    chunks = []

        if chunks:
            data.append({"items": chunks})

        return data

@RayAdmin anyone has some time to spare on this?

Can you provide a minimal repro script?