PySpark新手求助：处理近3个月分区数据并转存至S3

根据你的数据源分区结构，以下是一步步实现需求的具体代码和说明：

1. 初始化SparkSession #

创建SparkSession是PySpark程序的核心入口：

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

spark = SparkSession.builder \
    .appName("ProcessLast3MonthsData") \
    .getOrCreate()

2. 计算过去3个月的时间范围，过滤分区 #

从分区结构推断数据按year/month层级划分，先计算时间范围，再构造分区过滤条件避免全量扫描：

# 计算当前日期往前推3个月的起始点
end_date = datetime.now()
start_date = end_date - timedelta(days=90)

# 提取起止时间的年、月用于分区过滤
start_year, start_month = start_date.year, start_date.month
end_year, end_month = end_date.year, end_date.month

# 构造分区过滤条件（如果你的分区字段是date格式，需调整逻辑）
filter_condition = f"(year >= {start_year} AND month >= {start_month}) OR (year = {end_year} AND month <= {end_month})"

3. 读取数据源并筛选指定字段 #

读取时仅加载需要保留的字段，同时应用分区过滤：

# 替换为你的源数据S3路径
source_s3_path = "s3://your-source-bucket/path/to/data"

# 读取数据，只保留目标字段和分区字段
df = spark.read.parquet(source_s3_path) \
    .filter(filter_condition) \
    .select("old_field1", "old_field2", "old_field3", "year", "month")

4. 字段重命名 #

用withColumnRenamed批量或逐个完成字段重命名：

# 替换成你的实际字段映射关系
rename_mapping = {
    "old_field1": "new_field1",
    "old_field2": "new_field2",
    "old_field3": "new_field3"
}

# 批量执行重命名
for old_name, new_name in rename_mapping.items():
    df = df.withColumnRenamed(old_name, new_name)

5. 输出到目标S3桶，保留原分区结构 #

指定原分区字段，确保输出后的分区层级和源数据一致：

# 替换为你的目标S3路径
target_s3_path = "s3://your-target-bucket/path/to/output"

# 输出数据，根据业务需求选择模式（overwrite覆盖/append追加）
df.write.mode("overwrite") \
    .partitionBy("year", "month") \
    .parquet(target_s3_path)

关键注意事项 #

• 如果分区字段是date等其他格式，需调整时间计算和过滤逻辑。
• 读取时仅加载必要字段能大幅提升处理效率。
• 输出模式需根据业务场景选择，避免误删数据。

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