1

人们经常谈论使用parquet 和 pandas。而且我正在努力理解我们是否可以在与 pandas 一起使用时利用 parquet 文件的全部功能。例如，假设我有一个包含 30 列（包括年份、州、性别、姓氏）和许多行的大型镶木地板文件（按年份分区）。我想加载镶木地板文件并执行以下类似的计算

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 的原因）。

但我试图了解此查询在熊猫中的行为方式。它会在我们调用的那一刻将所有内容都带入记忆df = pd.read_parquet("file.parquet)吗？或者这里应用了任何惰性因素来引入投影和谓词下推？如果不是这种情况，那么在镶木地板上使用熊猫有什么意义呢？外面有这一切可能arrow package吗？

尽管我没有用过dask只是想知道这种情况是否在他们懒惰地执行时被处理。

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

4

1 回答 1

2

而且我正在努力理解我们是否可以在与 pandas 一起使用时利用 parquet 文件的全部功能。

TL;DR：是的，但你可能需要比使用 Dask 之类的东西更努力地工作。

例如说我有一个大的镶木地板文件（按年分区）

这是迂腐的，但单个镶木地板文件没有在任何东西上进行分区。Parquet“数据集”（文件集合）是分区的。例如：

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

在我们调用 df = pd.read_parquet("file.parquet) 的那一刻，它是否会将所有内容都带入内存？

是的。但是......你可以做得更好：

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

关键字会将过滤器向下传递给 pyarrow，它会以下filters推方式将过滤器应用到分区（例如，知道需要读取哪些目录）和行组统计信息。

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

使用箭头包可以实现这些吗？

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

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

以我的理解（我没有大量的 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 项目密切合作，因此我的回答可能偏向那个方向。

于 2022-03-04T02:14:53.987 回答