PySpark DataFrame按日期迭代筛选：每月取一周数据的实现求助

核心思路：避免循环，利用PySpark分布式特性批量处理 #

PySpark的分布式架构决定了循环迭代处理会频繁触发作业，效率极低。推荐直接通过日期函数+批量过滤或窗口函数实现需求，同时利用分区表特性优化性能。

方法一：基于日期范围批量过滤（推荐，适配分区表） #

该方法先生成每个月目标周的日期范围，再一次性过滤数据，能有效利用date分区减少扫描数据量。

步骤1：生成每月目标周的日期范围

这里以提取**每月第一周（周一至周日）**为例，可根据需求调整周起始日：

from pyspark.sql import functions as F
import pandas as pd

# 生成2018-01至2021-12的所有月份起始日
months = pd.date_range(start="2018-01-01", end="2021-12-31", freq="MS")

def get_target_week_range(month_start):
    # 获取当月第一天和最后一天
    first_day = month_start.date()
    month_last_day = (month_start + pd.offsets.MonthEnd(1)).date()
    
    # 计算当月第一个周一（若需周日起始，调整weekday判断逻辑）
    days_to_monday = (7 - first_day.weekday()) % 7
    week_start = first_day + pd.Timedelta(days=days_to_monday)
    # 周结束日为周日，不超过当月最后一天
    week_end = min(week_start + pd.Timedelta(days=6), month_last_day)
    
    return (week_start, week_end)

# 收集所有目标周的日期范围
target_ranges = [get_target_week_range(month) for month in months]

步骤2：构建过滤条件并读取数据

# 构建批量过滤表达式
filter_expr = F.lit(False)
for start, end in target_ranges:
    filter_expr = filter_expr | (F.col("date").between(start, end))

# 读取分区表并应用过滤（自动推分区，减少扫描）
df = spark.read.table("your_input_table").filter(filter_expr)

# 验证结果：按年月统计数据量
df.groupBy(F.year("date").alias("year"), F.month("date").alias("month")) \
  .count() \
  .orderBy("year", "month") \
  .show()

方法二：窗口函数筛选 #

如果需要灵活指定每月的第N周（比如第二周），可以用窗口函数按年月分区，对日期按周排序后筛选：

from pyspark.sql.window import Window

# 确保date字段为日期类型（若原字段是字符串需转换）
df = df.withColumn("date", F.to_date("date"))

# 按年、月分区，对日期按周排序，生成每月内的周序号
w = Window.partitionBy(F.year("date"), F.month("date")).orderBy(F.weekofyear("date"))
df_with_week_idx = df.withColumn("month_week_idx", F.dense_rank().over(w))

# 筛选每月第一周（改为2则取第二周，依此类推）
filtered_df = df_with_week_idx.filter(F.col("month_week_idx") == 1)

注意事项 #

• 周起始日：Spark的weekofyear函数默认以周日为一周起始，若需调整可通过spark.sql.session.timeZone或自定义日期逻辑修改。
• 分区优化：由于表按date分区，方法一的批量过滤会自动触发分区修剪，比窗口函数更高效。
• 数据类型：确保date字段是DateType，若为字符串需先用F.to_date()转换。

内容的提问来源于stack exchange，提问作者psiva