Spark持久化未达预期：UDF重复执行致API重复请求问题

问题根源 #

你遇到的核心问题是：调用persist()后，触发API请求的UDF仍被重复执行，导致collect()和merge()操作得到不同的Id值。主要原因有两点：

1. Spark DataFrame是不可变结构，persist()会返回一个带有持久化标记的新DataFrame，但你未将该返回值重新赋值给df，后续操作仍基于原始未持久化的DataFrame，每次action都会重新计算整个数据链路，包括触发API调用的UDF。
2. 默认的persist()存储级别为MEMORY_ONLY，若内存不足，缓存数据会被自动驱逐，同样会触发重新计算。

解决方案 #

1. 重新赋值持久化后的DataFrame并指定可靠存储级别 #

必须将persist()的结果赋值给df，同时选择DISK_ONLY或MEMORY_AND_DISK这类不易丢失的存储级别，确保数据持久化到磁盘。

2. 触发action完成缓存写入 #

调用persist()后，执行一个action（如count()）强制Spark完成数据缓存，避免后续操作才触发计算和缓存流程。

3. 后续操作统一使用持久化后的DataFrame #

修改后的代码示例：

from pyspark.storagelevel import StorageLevel

# 生成包含Id字段的DataFrame
df = df_source.withColumn(
    "Id",
    generateid_udf(col("ExID"), col("DueDate"), col("MoxiePDMID"), col("Details"), col("Name"))
)

# 持久化到磁盘，并重新赋值df
df = df.persist(StorageLevel.DISK_ONLY)

# 触发action，确保数据完成缓存
df.count()

# 后续操作使用缓存后的DataFrame
data = df.collect()
print(data)

# Merge操作
targetDF = DeltaTable.forPath(spark, targetPath)
(targetDF.alias("tgt")
 .merge(df.alias("src"), "tgt.ExID = src.ExID")
 .whenMatchedUpdate(
     set={
         "tgt.Id": "src.Id",
         "tgt.UpdatedBy": lit("Created"),
         "tgt.UpdatedDate": current_timestamp()
     }
 )
 .execute()
)

# 不再使用时释放缓存资源
df.unpersist()

补充说明 #

• 原代码失效原因：df.persist()未重新赋值，df仍为原始未标记持久化的DataFrame，collect()和merge()作为两次独立的action，会分别触发UDF执行和API调用，导致返回不同的Id。
• 存储级别选择：DISK_ONLY适合大数据集，避免内存不足导致缓存丢失；MEMORY_AND_DISK在内存充足时优先用内存，不足时写入磁盘，兼顾性能与可靠性。
• 资源释放：调用unpersist()可释放磁盘/内存资源，避免不必要的占用。

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