如何告诉 Spark 根据范围跳过分区

Posted 2023-03-23

【中文标题】如何告诉 Spark 根据范围跳过分区

【英文标题】：How to tell Spark to skip partitions based on a range

【发布时间】：2021-05-16 15:49:44

【问题描述】：

假设我有 100 个文件存储在 S3 中，属于我想使用 Spark SQL 查询的一个表。假设该表有一个 timestamp 列和其他一些列。时间戳是单调递增的，因此每个分区中的时间戳都是有序的，并且小于所有后续分区中的所有时间戳。

我现在想对这些文件进行查询，其中我有一个谓词timestamp between t1 and t2 或timestamp > t1，其中t1 可能是这100 个分区中间的时间戳。据我了解，Spark 目前无法确定时间戳在我的分区中是有序的，因此必须检查所有 100 个分区中的每一行并查看其时间戳。

我的问题是：如何让 Spark 变得更智能并跳过整个分区？例如，如果t1 在这 100 个分区的中间，我该如何教 Spark 跳过前 50 个分区？

我知道分区发现https://spark.apache.org/docs/latest/sql-data-sources-parquet.html#partition-discovery，但据我所知，这仅适用于“分类”分区，如gender=male，不适用于范围。我最好的选择是什么？

【问题讨论】：

你的数据格式是什么？是 Delta Lake、parquet 文件、csv 吗？

我们数据的格式是parquet

【参考方案1】：

我的情况基本相同，数据位于 /root/tenant=tenantId/date=2021-01-01 之类的文件夹中，我使用 spark 表达式（在 python 上）

from pyspark.sql.functions import to_date as fdate

events_df = spark.read.parquet('/root') \
        .filter(events_df.tenant_id == tenant_id) \
        .filter((from_date <= fdate(events_df.date, 'yyyy-MM-dd'))
                & (fdate(events_df.date, 'yyyy-MM-dd') <= to_date)) \
        .filter(col('item_name').isin(['val1', 'val2', 'val3'])) \
        .select(col('date').alias('file_date'),
                'device_id',
                'dataset_context_id',
                col('free').cast('float').alias('free'),
                'item_time')

我有以下执行计划：

== Physical Plan ==
*(1) Project [date#13 AS file_date#54, device_id#0, dataset_context_id#1, cast(free#9 as float) AS free#55, item_time#6]
+- *(1) Filter item_name#7 IN (val1,val2,val3)
   +- FileScan parquet [device_id#0,dataset_context_id#1,item_time#6,item_name#7,item_data#9,tenant_id#12,date#13] Batched: false, DataFilters: [item_name#7 IN (val1,val2,val3)], Format: Parquet, Location: InMemoryFileIndex(1 paths)[<path/to/root/>], PartitionFilters: [isnotnull(tenant_id#12), (tenant_id#12 = <tenantId>), (cast(cast(gettimestamp(date#13, yyyy-..., PushedFilters: [In(item_name, [val1,val2,val3])], ReadSchema: struct<device_id:string,dataset_context_id:string,item_time:timestamp,item_name:string,item_data:...

如你所见，计划中有PartitionFilters，这证明spark在日期范围内运行分区过滤。 （火花 3.1.1）