使用 pandas 和 parquet 的效率

Efficiency in using pandas and parquet

人们经常谈论使用 parquet and pandas。我正在努力了解我们是否可以在与 pandas 一起使用时利用 parquet 文件的全部功能。例如,假设我有一个大的镶木地板文件(按年份分区),其中包含 30 列(包括年份、州、性别、last_name)和许多行。我想加载 parquet 文件并执行类似的计算

import pandas as pd
df = pd.read_parquet("file.parquet")
df_2002 = df[df.year == 2002]
df_2002.groupby(["state", "gender"])["last_name"].count()

在此查询中,仅使用了 4 列(共 30 列)并且仅使用了年份 2002 分区。这意味着我们只想带来此计算所需的列和行,并且在带有谓词和投影下推的 parquet 中可以实现类似的事情(以及我们使用 parquet 的原因)。

但我试图了解此查询在 pandas 中的行为方式。它会在我们调用 df = pd.read_parquet("file.parquet) 的那一刻将所有内容带入内存吗?或者这里应用了任何惰性因素来引入投影和谓词下推?如果不是这种情况,那么将 pandas 与镶木地板一起使用有什么意义?使用 arrow package 可以实现这些吗?

尽管我没有使用过 dask 只是想知道这种情况是否会在 dask 中处理,因为他们懒惰地执行它。

我确信这种情况在 spark 世界中得到了很好的处理,但只是想知道在本地场景中如何使用 pandas、arrow、dask、ibis 等软件包处理这些情况

And I am trying hard to understand if we can utilize the entire features of parquet files when used with pandas.

TL;DR:是的,但与使用 Dask 之类的软件相比,您可能需要更加努力地工作。

For instance say I have a big parquet file (partitioned on year)

这是迂腐的,但单个镶木地板文件没有在任何地方分区。 Parquet“数据集”(文件集合)是分区的。例如:

my_dataset/year=2002/data.parquet
my_dataset/year=2003/data.parquet

Does it bring everything into memory the moment we call df = pd.read_parquet("file.parquet) ?

是的。但是……你可以做得更好:

df = pd.read_parquet('/tmp/new_dataset', filters=[[('year','=', 2002)]], columns=['year', 'state', 'gender', 'last_name'])

filters 关键字会将过滤器向下传递给 pyarrow,后者将以下推方式将过滤器应用于分区(例如,了解需要读取哪些目录)和行组统计信息。

columns 关键字会将列选择传递给 pyarrow,pyarrow 将应用选择以仅从磁盘读取指定的列。

Any of this is possible with the arrow package out there ?

pandas' read_parquet 文件中的所有内容都由 pyarrow 在幕后处理(除非您更改为其他引擎)。传统上,group_by 将由 pandas(好吧,也许是 numpy)直接处理,但是如果你想尝试在 pyarrow 中做所有事情,pyarrow 也有一些实验性计算 API。

Eventhough I haven't used dask just wondering if this kind of situation is handled in dask as they perform it lazily.

据我了解(我对 dask 没有太多经验),当你说...

df_2002 = df[df.year == 2002]
df_2002.groupby(["state", "gender"])["last_name"].count()

...在 dask 数据帧中,dask 会发现它可以应用下推过滤器和谓词,并且它会在加载数据时这样做。所以 dask 负责确定您应该应用哪些过滤器以及您需要加载哪些列。这样您就不必提前自己弄清楚了。

完整示例(您可以使用 strace 来验证它只加载两个镶木地板文件之一,并且只加载该文件的一部分):

import pyarrow as pa
import pyarrow.dataset as ds
import pandas as pd

import shutil

shutil.rmtree('/tmp/new_dataset')
tab = pa.Table.from_pydict({
    "year": ["2002", "2002", "2002", "2002", "2002", "2002", "2003", "2003", "2003", "2003", "2003", "2003"],
    "state": [ "HI",   "HI",   "HI",   "HI",   "CO",   "CO",   "HI",   "HI",   "CO",   "CO",   "CO",   "CO"],
    "gender": [ "M",    "F",   None,    "F",    "M",    "F",   None,    "F",    "M",    "F",    "M",    "F"],
 "last_name": ["Smi", "Will", "Stev", "Stan",  "Smi", "Will", "Stev", "Stan",  "Smi", "Will", "Stev", "Stan"],
    "bonus": [    0,      1,      2,      3,      4,      5,      6,      7,      8,      9,     10,     11]
})
ds.write_dataset(tab, '/tmp/new_dataset', format='parquet', partitioning=['year'], partitioning_flavor='hive')

df = pd.read_parquet('/tmp/new_dataset', filters=[[('year','=', 2002)]], columns=['year', 'state', 'gender', 'last_name'])
df_2002 = df[df.year == 2002]
print(df.groupby(["state", "gender"])["last_name"].count())

免责声明:您在这里询问的是多项技术。我与 Apache Arrow 项目密切合作,因此我的回答可能偏向于那个方向。