PySpark是否存在执行日期/运行日期参数？历史数据处理咨询

PySpark本身没有专门的"执行日期参数"来直接批量遍历指定日期范围，但可以通过生成目标日期序列的方式，轻松实现2022年历史数据的批量处理，以下是具体实现方案：

核心思路 #

针对2022年的每一天作为目标日期，动态计算该日期对应的过去10天数据窗口，完成计算后写入对应目录，完全替代实时场景下的datetime.now()逻辑。

方案一：Python日期生成 + PySpark遍历（适合小规模或本地调试） #

先通过Python生成2022年所有日期的序列，再逐个遍历处理：

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

# 初始化SparkSession（需配置ADLS Gen2访问权限）
spark = SparkSession.builder \
    .appName("HistoricalDataProcessing") \
    .config("fs.azure.account.auth.type", "OAuth") \
    .config("fs.azure.account.oauth.provider.type", "org.apache.hadoop.fs.azurebfs.oauth2.ClientCredsTokenProvider") \
    .config("fs.azure.account.oauth2.client.id", "<your-client-id>") \
    .config("fs.azure.account.oauth2.client.secret", "<your-client-secret>") \
    .config("fs.azure.account.oauth2.client.endpoint", "https://login.microsoftonline.com/<your-tenant-id>/oauth2/token") \
    .getOrCreate()

# 生成2022年所有日期
start_date = datetime(2022, 1, 1)
end_date = datetime(2022, 12, 31)
delta = timedelta(days=1)

current_date = start_date
while current_date <= end_date:
    # 计算当前目标日期对应的10天窗口
    target_date_str = current_date.strftime("%Y/%m/%d")
    window_start = current_date - timedelta(days=10)
    window_start_str = window_start.strftime("%Y/%m/%d")
    
    # 读取ADLS中过去10天的数据（使用通配符匹配日期目录）
    input_path = f"abfss://<container>@<storage-account>.dfs.core.windows.net/raw/{window_start_str}*/**"
    raw_data = spark.read.parquet(input_path)  # 根据实际数据格式调整read方法
    
    # 执行比率计算逻辑（替换为你的实际计算代码）
    ratio_df = raw_data.groupBy("some_column").agg(
        (sum("numerator") / sum("denominator")).alias("ratio")
    )
    
    # 写入目标目录
    output_path = f"abfss://<container>@<storage-account>.dfs.core.windows.net/changed/{target_date_str}"
    ratio_df.write.mode("overwrite").parquet(output_path)
    
    current_date += delta

方案二：全PySpark日期序列生成（适合集群分布式运行） #

用PySpark内置函数生成日期序列，避免Python单线程遍历的性能瓶颈：

from datetime import timedelta
from pyspark.sql import SparkSession
from pyspark.sql.functions import date_add, date_format, lit

spark = SparkSession.builder \
    .appName("DistributedHistoricalProcessing") \
    # 同上ADLS配置...
    .getOrCreate()

# 生成2022年所有日期的DataFrame
dates_df = spark.sql("""
    SELECT sequence(to_date('2022-01-01'), to_date('2022-12-31'), interval 1 day) as dates
""").selectExpr("explode(dates) as target_date")

# 定义处理单日期的逻辑
def process_date(target_date):
    target_date_str = target_date.strftime("%Y/%m/%d")
    window_start = target_date - timedelta(days=10)
    window_start_str = window_start.strftime("%Y/%m/%d")
    
    input_path = f"abfss://<container>@<storage-account>.dfs.core.windows.net/raw/{window_start_str}*/**"
    raw_data = spark.read.parquet(input_path)
    
    # 比率计算逻辑
    ratio_df = raw_data.groupBy("some_column").agg(
        (sum("numerator") / sum("denominator")).alias("ratio")
    )
    
    output_path = f"abfss://<container>@<storage-account>.dfs.core.windows.net/changed/{target_date_str}"
    ratio_df.write.mode("overwrite").parquet(output_path)
    return [target_date_str]

# 转换日期列并分布式处理
dates_df.rdd.map(lambda row: process_date(row.target_date)).collect()

关键优化点 #

• 分区读取优化：利用ADLS的目录结构，使用通配符*匹配日期分区，避免全表扫描。
• 小文件合并：写入时可设置.option("maxRecordsPerFile", 100000)合并小文件，提升后续查询性能。
• 增量处理：如果部分日期已处理，可先读取changed目录的已存在日期，过滤后只处理未完成的日期。

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