python - Azure 数据湖的 Pyarrow 切片下推

Question

我想访问 Azure 数据湖上的 Parquet 文件，并且仅检索某些行。

这是一个使用公共(public)数据集的可重现示例:

import pyarrow.dataset as ds
from adlfs import AzureBlobFileSystem

abfs_public = AzureBlobFileSystem(
    account_name="azureopendatastorage")

dataset_public = ds.dataset('az://nyctlc/yellow/puYear=2010/puMonth=1/part-00000-tid-8898858832658823408-a1de80bd-eed3-4d11-b9d4-fa74bfbd47bc-426339-18.c000.snappy.parquet', filesystem=abfs_public)

与收集完整数据集相比，收集 5 行的处理时间相同。有没有办法使用Pyarrow实现切片下推？

这是我的测试:

dataset_public.to_table()
# 5min 30s

dataset_public.head(5)
# 5min 11s

dataset_public.scanner().head(5)
# 5min 43s

我不确定 .head() 和 .scanner().head() 之间是否有区别

相关页面:

• Apache 箭头网站:https://arrow.apache.org/docs/python/parquet.html#reading-from-cloud-storage
• ADLFS Github 页面:https://github.com/fsspec/adlfs

Answer 1

经过一些调整，我想我得到了你想要的东西。首先，让我们看一下您发布的原始代码:

import pyarrow.dataset as ds
from adlfs import AzureBlobFileSystem

abfs_public = AzureBlobFileSystem(
    account_name="azureopendatastorage")

dataset_public = ds.dataset('az://nyctlc/yellow/puYear=2010/puMonth=1/part-00000-tid-8898858832658823408-a1de80bd-eed3-4d11-b9d4-fa74bfbd47bc-426339-18.c000.snappy.parquet', filesystem=abfs_public)

查看您提供的路径，您将其指向单个文件而不是整个数据集。添加一些调整:

import pyarrow.dataset as ds
from adlfs import AzureBlobFileSystem

abfs_public = AzureBlobFileSystem(account_name="azureopendatastorage")
dataset_public = ds.dataset('nyctlc/yellow/', filesystem=abfs_public, partitioning='hive')

现在，使用dataset_public.head(5)我得到:

由于我没有给它排序顺序，它只是抓取了它可以从恰好是第一个片段(最有可能)的任何文件中获取的前 5 行。

在您的原始代码示例中，您提供的路径使用 puYear=2010 和 puMonth=1，因此我们可以使用它们。因为我们告诉它使用 hive 分区，所以我们可以确认它发现数据集是根据这些值进行分区的:

print(dataset_public.partitioning.schema)
# prints:
# puYear: int32
# puMonth: int32
# -- schema metadata --
# org.apache.spark.sql.parquet.row.metadata: '{"type":"struct","fields":[{"' + 1456

如果我们使用这些字段作为过滤器获取前 5 行:

所以花了 1 分 31 秒。但我们可以做得更好!

W00t! 1.12秒

看，默认的batch_size相当大，我现在忘记它是什么了。但如果您只想获取少量行，您可以调整batch_size和片段预读大小等。以更好地适合您的用例。

如果您查看 head() 方法的基本 API 文档，它的 **kwargs 表示“请参阅 Scanner() 方法以获取完整参数描述” 。如果您转到 scanner() 方法，它会将您指向此处:https://arrow.apache.org/docs/python/generated/pyarrow.dataset.Scanner.html#pyarrow.dataset.Scanner.from_dataset，您可以在其中查看该方法的所有可用参数。就像只获取列的子集(因为 Parquet 非常高效):

我希望这可以帮助您更好地了解如何利用数据集 API 和粗糙边缘/技巧来提高性能。