我发现每当从 parquet 文件中读取数据帧时,dask.dataframe.rename 都会显着增加计算时间:
In [1]: import dask.dataframe as dd
...: df = dd.demo.daily_stock('GOOG', '2008', '2010', freq='1s', random_state=1234)
...:
In [2]: %time df.close.mean().compute()
CPU times: user 7.73 s, sys: 1.15 s, total: 8.88 s
Wall time: 3.5 s
Out[2]: 452.30345234893554
In [3]: %time df = df.rename(columns={col: col.upper() for col in df.columns}); df.CLOSE.mean().compute()
CPU times: user 8.06 s, sys: 1.21 s, total: 9.27 s
Wall time: 3.81 s
In [4]: df.to_parquet('df', compression='GZIP')
...: df = dd.read_parquet('df')
...:
In [5]: %time df.CLOSE.mean().compute()
CPU times: user 4.14 s, sys: 729 ms, total: 4.87 s
Wall time: 2.1 s
Out[5]: 452.30345234893554
In [6]: %time df = df.rename(columns={col: col.lower() for col in df.columns}); df.close.mean().compute()
CPU times: user 9.72 s, sys: 1.89 s, total: 11.6 s
Wall time: 4.81 s
请注意,原始数据帧上的差异很小,但基于 Parquet 的数据帧上的差异超过两倍。
这个问题在大型数据集 (~20-30GB) 上被夸大了,我看到 mean 计算从几秒到几分钟。
这是我不知道的 parquet 文件固有的东西,还是某种错误?
请您参考如下方法:
Parquet 是列式存储。从 parquet 文件中读取单个列比读取整个数据集要快得多。当你做 df.close.mean().compute() Dask 注意到你有一个 read_parquet操作后紧跟列访问操作,它可以智能地将它们融合到更智能的东西中,如下所示:
df = dd.read_parquet(filename, columns=['close'])
但是,当您抛出 rename read_parquet之间的操作调用和列访问操作 Dask.dataframe 不够智能,无法实现列访问和重命名之间的交换,因此您最终会从 parquet 文件中读取所有数据,重命名列,然后丢弃所有列除了一个。
缺乏执行有关计算的高级推理的能力正是数据库或更高级的系统(如 Spark Dataframes)开始胜过 Dask.dataframe 的地方。 Dask 的核心通常是较低级别的,因此可以进行更疯狂的计算,但除了最基本的查询优化外,它失去了执行任何操作的能力。
所以在这种情况下,rename 不是事实正在放慢速度,这是rename的事实正在对一个非常简单的优化方案进行改进。
