3

我正在生成一个 Dask 数据帧，以在 dask-ml 提供的聚类算法中下游使用。在我管道的上一步中，我使用 读取磁盘中的数据帧，使用dask.dataframe.read_parquet应用转换来添加列map_partitions，然后使用 将生成的数据帧写回磁盘dask.dataframe.to_parquet。再次读取并compute()调用生成的数据帧时会出现此问题。

运行以下代码：

# First step: make the Dask dataframe
ddf = ddf.map_partitions(partition_func)  # in this case, perform a df.apply, then pandas.concat with the original
ddf_output_path = pathlib.Path("./data/")  # Some directory
ddf_output_path.mkdir(parents=True, exist_ok=True)
dask.dataframe.to_parquet(ddf, ddf_output_path)  # Succeeds

# Second step: attempt to read and compute on the Dask dataframe
ddf = dask.dataframe.read_parquet(ddf_output_path)
print(ddf.columns)  # Produces the correct output
print(ddf.shape[0].compute())  # <-- fails here, for example

产生以下回溯：

 File "/home/ec2-user/pycharm_remote/pipeline/perform_clustering.py", line 32, in run
    print(ddf.shape[0].compute())
  File "/home/ec2-user/project/virtualenv/lib64/python3.7/site-packages/dask/base.py", line 167, in compute
    (result,) = compute(self, traverse=False, **kwargs)
  File "/home/ec2-user/project/virtualenv/lib64/python3.7/site-packages/dask/base.py", line 452, in compute
    results = schedule(dsk, keys, **kwargs)
  File "/home/ec2-user/project/virtualenv/lib64/python3.7/site-packages/dask/threaded.py", line 84, in get
    **kwargs
  File "/home/ec2-user/project/virtualenv/lib64/python3.7/site-packages/dask/local.py", line 486, in get_async
    raise_exception(exc, tb)
  File "/home/ec2-user/project/virtualenv/lib64/python3.7/site-packages/dask/local.py", line 316, in reraise
    raise exc
  File "/home/ec2-user/project/virtualenv/lib64/python3.7/site-packages/dask/local.py", line 222, in execute_task
    result = _execute_task(task, data)
  File "/home/ec2-user/project/virtualenv/lib64/python3.7/site-packages/dask/core.py", line 121, in _execute_task
    return func(*(_execute_task(a, cache) for a in args))
  File "/home/ec2-user/project/virtualenv/lib64/python3.7/site-packages/dask/dataframe/io/parquet/core.py", line 276, in read_parquet_part
    dfs = [func(fs, rg, columns.copy(), index, **kwargs) for rg in part]
  File "/home/ec2-user/project/virtualenv/lib64/python3.7/site-packages/dask/dataframe/io/parquet/core.py", line 276, in <listcomp>
    dfs = [func(fs, rg, columns.copy(), index, **kwargs) for rg in part]
  File "/home/ec2-user/project/virtualenv/lib64/python3.7/site-packages/dask/dataframe/io/parquet/arrow.py", line 758, in read_partition
    piece, columns, partitions, **kwargs
  File "/home/ec2-user/project/virtualenv/lib64/python3.7/site-packages/dask/dataframe/io/parquet/arrow.py", line 817, in _parquet_piece_as_arrow
    **kwargs.get("read", {}),
  File "/home/ec2-user/project/virtualenv/lib64/python3.7/site-packages/pyarrow/parquet.py", line 719, in read
    table = reader.read_row_group(self.row_group, **options)
  File "/home/ec2-user/project/virtualenv/lib64/python3.7/site-packages/pyarrow/parquet.py", line 272, in read_row_group
    use_threads=use_threads)
  File "pyarrow/_parquet.pyx", line 1080, in pyarrow._parquet.ParquetReader.read_row_group
  File "pyarrow/_parquet.pyx", line 1099, in pyarrow._parquet.ParquetReader.read_row_groups
  File "pyarrow/error.pxi", line 99, in pyarrow.lib.check_status
OSError: Couldn't deserialize thrift: TProtocolException: Invalid data
Deserializing page header failed.

环境是 Amazon Linux 2，Python 3.7.9，dask == 2.30.0，pyarrow == 2.0.0，pandas == 1.1.5，numpy == 1.19.4。dask 数据帧由 404 列组成，从大约 14,000 个 parquet 文件（分区）中读取。其中四列包含 type 项object（三列包含字符串，一列包含字符串的嵌套列表），其余 400 列包含 type float64。

4

0 回答 0