Spark 3.3创建Parquet分区并每日更新数据及合并文件问题

针对你的需求（按profile分区、单分区单Parquet文件、profile+date维度数据更新），以下是具体实现步骤：

1. 读取当日增量CSV数据 #

手动指定Schema以提升性能并避免类型推断错误：

from pyspark.sql import SparkSession
from pyspark.sql.types import StructType, StructField, StringType, DateType, DoubleType

spark = SparkSession.builder.appName("ParquetDailyUpdate").getOrCreate()

# 定义数据Schema
schema = StructType([
    StructField("profile", StringType(), nullable=True),
    StructField("date", DateType(), nullable=True),
    StructField("local_timezone", StringType(), nullable=True),
    StructField("utility", StringType(), nullable=True),
    StructField("iso", StringType(), nullable=True),
    StructField("unit", StringType(), nullable=True)
] + [StructField(f"hour_{i}", DoubleType(), nullable=True) for i in range(24)])

# 读取当日CSV（替换为你的实际S3路径）
new_df = spark.read.csv(
    "s3://your-input-bucket/daily-files/20240520/*.csv",
    header=True,
    schema=schema
)

2. 读取已有分区数据 #

处理首次运行无数据的场景：

try:
    existing_df = spark.read.parquet("s3://your-output-bucket/parquet-data/")
except Exception:
    # 首次运行时初始化空DataFrame
    existing_df = spark.createDataFrame([], schema=new_df.schema)

3. 合并数据：覆盖重复的profile+date行 #

通过左反连接过滤掉已有数据中与新数据重复的profile+date行，再合并新数据，实现旧数据替换：

# 保留现有数据中未被新数据覆盖的行
filtered_existing = existing_df.join(
    new_df,
    on=["profile", "date"],
    how="left_anti"
)

# 合并过滤后的旧数据与新数据
merged_df = filtered_existing.union(new_df)

4. 重分区确保单分区单文件 #

按profile重分区，让每个profile的数据集中在一个Spark分区，写入时自动生成单个Parquet文件：

# 按profile重分区，每个profile对应一个Spark分区
single_file_df = merged_df.repartition("profile")

注意：若profile数量极多（如10k+），此方式会引发Shuffle性能问题，需调整为按profile哈希分区并控制总分区数，或后续单独合并小文件。

5. 配置Spark参数并写入Parquet #

启用动态分区覆盖，避免误删未更新的分区，同时适配S3对象存储特性：

# 开启动态分区覆盖（仅覆盖有更新的profile分区）
spark.conf.set("spark.sql.sources.partitionOverwriteMode", "dynamic")
spark.conf.set("hive.exec.dynamic.partition", "true")
spark.conf.set("hive.exec.dynamic.partition.mode", "nonstrict")

# S3专属优化：解决对象存储原子性与冲突问题
spark.conf.set("spark.hadoop.fs.s3a.committer.name", "directory")
spark.conf.set("spark.hadoop.fs.s3a.committer.staging.conflict-mode", "replace")

# 写入数据（overwrite模式配合动态分区覆盖，仅更新目标分区）
single_file_df.write.partitionBy("profile")\
    .mode("overwrite")\
    .parquet("s3://your-output-bucket/parquet-data/")

核心逻辑说明 #

• 动态分区覆盖：spark.sql.sources.partitionOverwriteMode=dynamic 确保仅覆盖有新数据写入的profile分区，不会影响其他未更新的分区。
• 单文件控制：repartition("profile") 让每个profile的数据进入同一个Spark任务，写入时生成单个Parquet文件。
• 数据更新：通过left_anti连接过滤旧数据，合并新数据，实现profile+date维度的全量替换。

额外建议 #

• 幂等性保障：记录已处理的CSV文件名至数据库（如DynamoDB），避免重复运行ETL时重复生成数据。
• Schema一致性：每日CSV的Schema需与Parquet Schema保持一致，若有变更需提前做Schema兼容处理。
• 大文件优化：若数据量极大，可先按profile分区后，再对单个分区执行coalesce(1)进一步合并文件。

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