简化按分区读取Delta表历史7天数据的条件实现咨询

嘿，关于你这个按year/month/day（字符串类型）分区的Delta表读取需求，我有个简洁且能利用分区 pruning 的方案，完全不用拼接分区列就能实现精准过滤，一起来看看吧！

首先先明确你的需求：当任务在周二运行时，要取前一周的周一到周日共7天数据（比如触发日期是2024-05-07周二，就取2024-04-29到2024-05-05的7天）。核心痛点是不想拼接分区列成日期字符串，直接用三个分区字段过滤，同时保证只扫描目标分区。

核心思路 #

我们可以先生成需要覆盖的所有7天的日期列表，然后把每个日期拆成对应字符串格式的year/month/day，最后用Spark的isin条件直接匹配这三个分区字段的组合——这样Spark会自动识别分区过滤规则，只读取符合条件的分区，不会做全表扫描。

完整代码实现 #

from datetime import datetime, timedelta
from pyspark.sql.functions import col
from pyspark.sql import SparkSession

# 初始化SparkSession（如果你的环境已经初始化过可以跳过）
spark = SparkSession.builder.appName("Delta7DaysData").getOrCreate()

# 触发日期（可以替换成动态获取的当前日期，比如datetime.today().strftime('%Y-%m-%d')）
trigger_date = "2024-05-07"
trigger_date_obj = datetime.strptime(trigger_date, '%Y-%m-%d')

# 计算日期范围：前8天到前2天（包含两端，刚好7天）
start_date_obj = trigger_date_obj - timedelta(days=8)
end_date_obj = trigger_date_obj - timedelta(days=2)

# 生成这个区间内的所有日期
date_range = []
current_date = start_date_obj
while current_date <= end_date_obj:
    date_range.append(current_date)
    current_date += timedelta(days=1)

# 把每个日期转换成(年字符串, 月字符串(补前导零), 日字符串(补前导零))的元组
# 因为你的分区列是字符串类型，必须和分区目录的格式一致（比如5月要写成"05"而非"5"）
partition_filters = [
    (
        str(date.year),
        f"{date.month:02d}",
        f"{date.day:02d}"
    )
    for date in date_range
]

# 读取Delta表并应用分区过滤
df = spark.read.format("delta") \
    .load("/path/to/your/delta/table") \
    .filter(
        (col("year"), col("month"), col("day")).isin(partition_filters)
    )

# 可以验证一下读取的分区是否正确
df.select("year", "month", "day").distinct().show()

方案优势 #

• 高效利用分区 pruning：Spark会直接根据year/month/day的组合过滤分区，只读取目标目录的数据，性能比拼接字符串过滤好很多。
• 代码简洁易维护：不用处理跨月/跨年的复杂分支判断，生成日期列表的逻辑一目了然。
• 避免格式错误：通过补前导零保证分区字段的格式和实际分区目录完全匹配，不会出现因为"5"和"05"不匹配导致的漏读/错读。

如果你的触发日期是动态获取的（比如任务运行当天），只需要把trigger_date改成datetime.today().strftime('%Y-%m-%d')就可以啦！

备注：内容来源于stack exchange，提问作者Matthew