Spark读写同一S3路径报无法推断Schema错误，如何优化Upsert跳过临时路径

问题原因 #

报错的核心是读写同一路径的冲突：Spark的cache()是懒执行操作，调用缓存后直接执行写入时，缓存还没有被实际物化。Spark执行overwrite写入前会先清空目标S3路径，等到后续计算流程需要读取原old_data的数据时，原文件已经被删除，因此触发报错。

可行解决方案 #

方案1：手动触发缓存物化（无框架改造，适合小集群） #

在缓存后、写入前执行全量触发计算的action，确保final_data已经完全缓存到内存/本地磁盘，不再依赖原S3路径的文件：

from pyspark import StorageLevel

# 选用MEMORY_AND_DISK_SER存储级别，内存不足时自动溢写磁盘，避免内存溢出
final_data.persist(StorageLevel.MEMORY_AND_DISK_SER)
# 触发全量计算完成缓存物化，count是最简单的全量触发操作
final_data.count()
# 此时写入不会再读取原old_data的S3路径
final_data.write.mode("overwrite").parquet('s3://bucket/old_data/')
# 写入完成后手动释放缓存
final_data.unpersist()

注意：该方案需要集群可用内存+本地磁盘容量能够容纳全量final_data，否则会出现溢写性能下降或者OOM问题。

方案2：使用湖仓存储格式（推荐，十亿级数据性能最优） #

使用Delta Lake、Apache Iceberg等支持ACID的湖仓格式替换原生Parquet，内置Upsert能力，不需要手写join逻辑，也不会出现读写冲突。
以Delta Lake为例，操作代码如下：

from delta.tables import DeltaTable

# 首次使用时将原Parquet数据转成Delta格式，仅需执行一次
# old_data.write.format("delta").save('s3://bucket/old_data_delta/')

delta_table = DeltaTable.forPath(spark, 's3://bucket/old_data_delta/')
# 直接执行merge upsert操作，无需全量读取重写旧数据
delta_table.alias("old") \
  .merge(
    new_data.alias("new"),
    "old.opk = new.opk"
  ) \
  .whenMatchedUpdateAll() \
  .whenNotMatchedInsertAll() \
  .execute()

该方案仅更新匹配主键的对应数据，无需全量重写旧表，十亿级数据下性能比原生join+全量重写高10倍以上。

现有逻辑优化提示 #

你当前伪代码中的common_records是新旧数据inner join的结果，直接union会出现字段重复的问题，需要调整为取新数据的对应字段，才能实现新数据覆盖旧数据的逻辑：

common_records = new_data.join(old_data, on=opk, how="inner")

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