Spark Checkpoint后如何正确重新计算DataFrame统计信息？

针对Checkpoint后缺失统计信息的问题，以下是几种实用的处理方式：

1. 手动执行ANALYZE TABLE计算统计 #

Checkpoint后通过临时表触发统计计算，这是最直接的修复方式：

finalDF = finalDF.checkpoint()
// 注册临时表
finalDF.createOrReplaceTempView("temp_iter_df")
// 计算全列统计信息
spark.sql("ANALYZE TABLE temp_iter_df COMPUTE STATISTICS FOR ALL COLUMNS")
// 重新获取带统计的DataFrame
finalDF = spark.table("temp_iter_df")

该操作会让Spark扫描Checkpoint后的实际数据，生成准确的统计信息，优化器能基于真实数据量生成合理执行计划。

2. 缓存+Checkpoint结合保留统计 #

先缓存数据并触发计算（自动生成统计），再进行Checkpoint截断血统：

finalDF = finalDF.join(finalDF, Seq("UniqueCarrier"))
// 缓存数据并触发落地
finalDF.cache()
finalDF.count() // 触发缓存的同时，Spark会自动计算统计信息
// 执行Checkpoint截断血统
finalDF = finalDF.checkpoint()

缓存阶段生成的统计信息会被保留到Checkpoint后的DataFrame中，避免优化器使用失真估算。

3. 配置Spark自动计算统计 #

通过以下配置让Spark在Checkpoint后自动尝试统计刷新：

// 允许从文件系统获取文件大小作为统计参考
spark.conf.set("spark.sql.statistics.fallBackToHdfs", true)
// 自动计算数据大小统计
spark.conf.set("spark.sql.statistics.size.autoCompute", true)

该配置适合基于文件存储的Checkpoint场景，Spark会自动从DBFS/HDFS读取Checkpoint文件的大小，结合数据分布生成更合理的统计估算。

4. 临时表查询+Checkpoint间接保留统计 #

如果不需要绝对严格的血统截断，可通过临时表查询生成带统计的逻辑计划，再进行Checkpoint：

finalDF = finalDF.join(finalDF, Seq("UniqueCarrier"))
finalDF.createOrReplaceTempView("iter_df")
// 查询临时表生成带统计的计划，再Checkpoint
finalDF = spark.sql("SELECT * FROM iter_df").checkpoint()

这种方式既能截断主要血统，又能保留临时表查询阶段生成的统计信息。

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