使用Dask操作Parquet时遇TypeError问题求助

问题描述 #

通过conda install dask安装Dask 2022.7.0版本后，操作Parquet文件时持续触发TypeError("read_row_group_file() got an unexpected keyword argument 'infile'")异常，无论调用何种compute函数都会出现该问题。将Dask版本降级至2022.6.1后问题解决，现确认该问题属于软件版本兼容bug，而非操作失误。

报错堆栈 #

File ~\Desktop\prog\kaggle\amex\to_parquet.py:83 in <module>
    memory_usage["raw"] = df1.memory_usage(deep=True).sum().compute()/1048576

  File ~\anaconda3\envs\amex\lib\site-packages\dask\base.py:315 in compute
    (result,) = compute(self, traverse=False, **kwargs)

  File ~\anaconda3\envs\amex\lib\site-packages\dask\base.py:598 in compute
    results = schedule(dsk, keys, **kwargs)

  File ~\anaconda3\envs\amex\lib\site-packages\distributed\client.py:3001 in get
    results = self.gather(packed, asynchronous=asynchronous, direct=direct)

  File ~\anaconda3\envs\amex\lib\site-packages\distributed\client.py:2175 in gather
    return self.sync(

  File ~\anaconda3\envs\amex\lib\site-packages\distributed\utils.py:338 in sync
    return sync(

  File ~\anaconda3\envs\amex\lib\site-packages\distributed\utils.py:405 in sync
    raise exc.with_traceback(tb)

  File ~\anaconda3\envs\amex\lib\site-packages\distributed\utils.py:378 in f
    result = yield future

  File ~\anaconda3\envs\amex\lib\site-packages\tornado\gen.py:762 in run
    value = future.result()

  File ~\anaconda3\envs\amex\lib\site-packages\distributed\client.py:2038 in _gather
    raise exception.with_traceback(traceback)

  File ~\anaconda3\envs\amex\lib\site-packages\dask\optimization.py:990 in __call__
    return core.get(self.dsk, self.outkey, dict(zip(self.inkeys, args)))

  File ~\anaconda3\envs\amex\lib\site-packages\dask\core.py:149 in get
    result = _execute_task(task, cache)

  File ~\anaconda3\envs\amex\lib\site-packages\dask\core.py:119 in _execute_task
    return func(*(_execute_task(a, cache) for a in args))

  File ~\anaconda3\envs\amex\lib\site-packages\dask\core.py:119 in <genexpr>
    return func(*(_execute_task(a, cache) for a in args))

  File ~\anaconda3\envs\amex\lib\site-packages\dask\core.py:113 in _execute_task
    return [_execute_task(a, cache) for a in arg]

  File ~\anaconda3\envs\amex\lib\site-packages\dask\core.py:113 in <listcomp>
    return [_execute_task(a, cache) for a in arg]

  File ~\anaconda3\envs\amex\lib\site-packages\dask\core.py:119 in _execute_task
    return func(*(_execute_task(a, cache) for a in args))

  File ~\anaconda3\envs\amex\lib\site-packages\dask\dataframe\io\parquet\core.py:89 in __call__
    return read_parquet_part(

  File ~\anaconda3\envs\amex\lib\site-packages\dask\dataframe\io\parquet\core.py:587 in read_parquet_part
    dfs = [

  File ~\anaconda3\envs\amex\lib\site-packages\dask\dataframe\io\parquet\core.py:588 in <listcomp>
    func(fs, rg, columns.copy(), index, **toolz.merge(kwargs, kw))

  File ~\anaconda3\envs\amex\lib\site-packages\dask\dataframe\io\parquet\fastparquet.py:1001 in read_partition
    return cls.pf_to_pandas(

  File ~\anaconda3\envs\amex\lib\site-packages\dask\dataframe\io\parquet\fastparquet.py:1093 in pf_to_pandas
    pf.read_row_group_file(

TypeError: read_row_group_file() got an unexpected keyword argument 'infile'

相关代码片段 #

blocksize="200mb"
memory_usage = {}
df = ddf.read_csv(
    csv_path,
    blocksize=blocksize).sample(random_state=random_seed, frac=sample_size)

ddf.to_parquet(
    df=df, 
    path=raw_data_path,
    write_index=False,
    )
print("Data saved to:", raw_data_path)

df1 = ddf.read_parquet(raw_data_path)
memory_usage["raw"] = df1.memory_usage(deep=True).sum().compute()/1048576
print("Data size: ", df1.size.compute())
print("Current total memory usage: ",memory_usage["raw"])

相关依赖列表 #

regex-2022.7.9       | 307 KB    
dask-2022.7.0        | 20 KB     
holoviews-1.15.0     | 4.0 MB    
tenacity-8.0.1       | 34 KB     
bokeh-2.4.3          | 7.6 MB    
pip-22.1.2           | 2.5 MB    
qtconsole-5.3.1      | 201 KB    
zeromq-4.3.4         | 4.2 MB    
distributed-2022.7.0 | 1.1 MB    
requests-2.28.1      | 99 KB     
libsodium-1.0.18     | 477 KB    
pandas-1.4.3         | 8.7 MB    
ca-certificates-2022 | 123 KB    
libxgboost-1.5.0     | 1.3 MB    
ujson-5.4.0          | 44 KB     
notebook-6.4.11      | 4.5 MB    
pyct-0.4.8           | 45 KB     
py-xgboost-1.5.0     | 171 KB    
xgboost-1.5.0        | 26 KB     
ipython-8.4.0        | 1009 KB   
numpydoc-1.4.0       | 86 KB     
pillow-9.2.0         | 908 KB    
dask-core-2022.7.0   | 1.7 MB    
sphinx-5.0.2         | 1.7 MB    
_py-xgboost-mutex-2. | 12 KB     
docutils-0.18.1      | 666 KB    
panel-0.13.1         | 11.0 MB   
pyzmq-23.2.0         | 404 KB    

问题原因分析 #

该错误是Dask 2022.7.0与fastparquet版本不兼容导致的软件bug：

• Dask 2022.7.0对Parquet读取逻辑进行了更新，在调用fastparquet的read_row_group_file方法时新增了infile参数；
• 但当前环境中的fastparquet版本未同步支持该参数，从而触发参数不匹配的TypeError；
• 降级至Dask 2022.6.1后问题解决，是因为旧版本Dask未引入该参数，与现有fastparquet版本兼容。

内容的提问来源于stack exchange，提问作者Tormod