3

我正在使用 dask 来编写和阅读镶木地板。我正在使用 fastparquet 引擎写作并使用 pyarrow 引擎阅读。我的工人有 1 GB 的内存。使用 fastparquet 内存使用情况很好，但是当我切换到 pyarrow 时，它就会爆炸并导致工作人员重新启动。我在下面有一个可重现的示例，它在 1gb 内存限制的工作人员上使用 pyarrow 失败。实际上，我的数据集比这大得多。使用 pyarrow 的唯一原因是，与 fastparquet 相比，它在扫描时提高了我的速度（大约 7x-8x）

黄昏：0.17.1

pyarrow：0.9.0.post1

快速镶木地板：0.1.3

import dask.dataframe as dd
import numpy as np
import pandas as pd

size = 9900000
tmpdir = '/tmp/test/outputParquet1'

d = {'a': np.random.normal(0, 0.3, size=size).cumsum() + 50,
    'b': np.random.choice(['A', 'B', 'C'], size=size),
    'c': np.random.choice(['D', 'E', 'F'], size=size),
    'd': np.random.normal(0, 0.4, size=size).cumsum() + 50,
    'e': np.random.normal(0, 0.5, size=size).cumsum() + 50,
    'f': np.random.normal(0, 0.6, size=size).cumsum() + 50,
    'g': np.random.normal(0, 0.7, size=size).cumsum() + 50}
df = dd.from_pandas(pd.DataFrame(d), 200)
df.to_parquet(tmpdir, compression='snappy', write_index=True, 
         engine='fastparquet')

#engine = 'pyarrow' #fails due to worker restart
engine = 'fastparquet' #works fine
df_partitioned = dd.read_parquet(tmpdir + "/*.parquet", engine=engine)
print(df_partitioned.count().compute())
df_partitioned.query("b=='A'").count().compute()

编辑：我的原始设置运行 spark 作业，使用 fastparquet 将数据并行写入分区。因此元数据文件是在最里面的分区而不是父目录中创建的。因此使用 glob 路径而不是父目录（fastparquet 读取父目录时要快得多，而使用 glob 路径扫描时 pyarrow 获胜）

4

3 回答 3

0

我建议在read_parquet通话中选择您需要的列

df = dd.read_parquet('/path/to/*.parquet', engine='pyarrow', columns=['b'])

这将允许您有效地只读取您需要的几列，而不是一次读取所有列。

于 2018-06-15T11:47:37.130 回答
0

您还可以在加载数据时进行过滤。例如，按特定列进行过滤 df = dd.read_parquet('/path/to/*.parquet', engine='fastparquet', filters=[(COLUMN, 'operation', 'SOME_VALUE')])。

想象一下像==, >, <, 等等这样的操作。

于 2019-09-16T14:53:55.640 回答
0

我的非内存限制系统上的一些计时结果

使用您的示例数据

In [17]: df_partitioned = dd.read_parquet(tmpdir, engine='fastparquet')

In [18]: %timeit df_partitioned.count().compute()
2.47 s ± 114 ms per loop (mean ± std. dev. of 7 runs, 1 loop each)

In [19]: df_partitioned = dd.read_parquet(tmpdir, engine='pyarrow')

In [20]: %timeit df_partitioned.count().compute()
1.93 s ± 96.9 ms per loop (mean ± std. dev. of 7 runs, 1 loop each)

使用列b并c在写入之前转换为分类

In [30]: df_partitioned = dd.read_parquet(tmpdir, engine='fastparquet')

In [31]: %timeit df_partitioned.count().compute()
1.25 s ± 83.7 ms per loop (mean ± std. dev. of 7 runs, 1 loop each)

In [32]: df_partitioned = dd.read_parquet(tmpdir, engine='pyarrow')

In [33]: %timeit df_partitioned.count().compute()
1.82 s ± 63 ms per loop (mean ± std. dev. of 7 runs, 1 loop each)

使用 fastparquet 直接，单线程

In [36]: %timeit fastparquet.ParquetFile(tmpdir).to_pandas().count()
1.82 s ± 19 ms per loop (mean ± std. dev. of 7 runs, 1 loop each)

使用 20 个分区而不是 200 个（fastparquet、类别）

In [42]: %timeit df_partitioned.count().compute()
863 ms ± 78.8 ms per loop (mean ± std. dev. of 7 runs, 1 loop each)
于 2018-06-15T14:28:39.113 回答