我如何确保在查询分区数据框时读取不必要的文件的dask？

发布于 2025-02-14 00:00:06 字数 643 浏览 1 评论 0原文

例如，说我在目录树中有一个镶木quet数据集，hive style：

dataset/field1=A/field2=X/data.parquet
dataset/field1=A/field2=Y/data.parquet
dataset/field1=B/field2=X/data.parquet
dataset/field1=B/field2=Y/data.parquet

并说我定义了这样的dask dataframe：

import dask.dataframe as dd

df = dd.read_parquet("dataset/", engine="pyarrow")

here，field1 and field2 是分区列在我的数据框中。

我的问题：如果我执行df.query（“ field1 =='a'an and field2 =='x'”），我该如何确保Dask仅读取一个Parquet文件？（即，仅读取dataset/field1 = a/field2 = x/data.parquet）

是否有一种方法可以知道哪些文件dask正在打开和读取？也许是从记录仪中？

原文

For example, say I have a parquet dataset in a directory tree, hive-style:

dataset/field1=A/field2=X/data.parquet
dataset/field1=A/field2=Y/data.parquet
dataset/field1=B/field2=X/data.parquet
dataset/field1=B/field2=Y/data.parquet

And say I define a dask dataframe like this:

import dask.dataframe as dd

df = dd.read_parquet("dataset/", engine="pyarrow")

Here, field1 and field2 are partition columns in my dataframe.

My question: how do I make sure dask only reads one parquet file if I do df.query("field1 == 'A' and field2 == 'X'")? (i.e., only read dataset/field1=A/field2=X/data.parquet)

Relatedly, is there a way of knowing which files dask is opening and reading? Maybe from a logger?

分享到QQ

分享到微博

如果你对这篇内容有疑问，欢迎到本站社区发帖提问参与讨论，获取更多帮助，或者扫码二维码加入 Web 技术交流群。

发布评论

需要登录才能够评论，你可以免费注册一个本站的账号。

回首观望 2025-02-21 00:00:06

此功能由pyarrow在阅读时间使用过滤器时（不是.Query方法）：

from dask.dataframe import read_parquet
from dask.datasets import timeseries

df = timeseries(end="2000-01-03", seed=0).reset_index()
df["date"] = df["timestamp"].dt.date.astype("str")
df.to_parquet("test.pqt", partition_on=["id", "date"])
rare_id = df["id"].value_counts().compute().tail(1).index.values[0]

ddf = read_parquet("test.pqt", filters=[[("id", "==", rare_id)]], engine="pyarrow")
print(ddf.npartitions)  # 1

编辑：上面的片段与dask = 2022.1.0 <一起使用。 /code>，但不是dask = 2022.6.1，有关详细信息检查这个GH问题。

This functionality is supported by pyarrow when using filters at read time (not as .query method):

from dask.dataframe import read_parquet
from dask.datasets import timeseries

df = timeseries(end="2000-01-03", seed=0).reset_index()
df["date"] = df["timestamp"].dt.date.astype("str")
df.to_parquet("test.pqt", partition_on=["id", "date"])
rare_id = df["id"].value_counts().compute().tail(1).index.values[0]

ddf = read_parquet("test.pqt", filters=[[("id", "==", rare_id)]], engine="pyarrow")
print(ddf.npartitions)  # 1

Edit: the snippet above works with dask=2022.1.0, but not dask=2022.6.1, for details check this GH issue.

回复收藏 0 原文

~没有更多了~