PySpark中第二次执行Insert Overwrite时数据计数不匹配问题

我之前在使用PySpark 2.x处理分区表时也碰到过类似的计数不一致问题，结合你的代码来看，大概率和动态分区覆盖的默认行为、临时视图的状态或者DataFrame缓存残留有关，下面给你一步步排查和解决的方案：

1. 调整动态分区覆盖配置（最可能的根因） #

在PySpark 2.1.1中，默认的spark.sql.sources.partitionOverwriteMode参数值是static，这个模式下，如果你用动态分区插入（也就是SQL里不指定具体的data_date和data_product值，靠select语句自动推断分区），行为会不符合预期：

• 第一次运行时能正常写入所有分区；
• 第二次运行时，可能不会完全覆盖已存在的分区，要么残留旧数据，要么错误追加，最终导致总行数和预期不符。

解决方法：
执行insert语句前，把这个参数改成dynamic，让Spark只覆盖select语句涉及到的分区：

spark.conf.set("spark.sql.sources.partitionOverwriteMode", "dynamic")
fact = spark.sql("insert overwrite table " + usageWideFactTable + " partition (data_date, data_product) select * from usage_fact_staging")

2. 确保临时视图每次都是全新生成的 #

虽然你用了createOrReplaceTempView，但如果之前的usage_fact DataFrame被Spark自动缓存（比如多次触发action操作后），第二次运行时可能不会重新计算union和repartition逻辑，导致临时视图还是旧数据。

解决方法：
创建临时视图前主动清除缓存：

# 解除DataFrame的缓存（如果存在）
usage_fact.unpersist()
usage_fact.createOrReplaceTempView("usage_fact_staging")

也可以提前删除临时视图（虽然createOrReplace会替换，但多一层保险）：

if spark.catalog._jcatalog.tableExists("usage_fact_staging"):
    spark.sql("DROP VIEW usage_fact_staging")
usage_fact.createOrReplaceTempView("usage_fact_staging")

3. 确认Union的数据源每次都重新加载 #

你的usage_fact是多个DataFrame的union结果，要确保第二次运行时，gtac_usage、gtp_usage、upaf_src这些数据源都是重新读取的，而非使用缓存的旧数据。

解决方法：
在生成这些数据源的代码后加上unpersist()：

gtac_usage = spark.read.table("gtac_usage_table").unpersist()
gtp_usage = spark.read.table("gtp_usage_table").unpersist()
upaf_src = spark.read.table("upaf_src_table").unpersist()

4. 避免使用select *，明确指定列名 #

PySpark 2.1.1中，insert overwrite ... partition(...)要求select语句的列顺序、数量和目标表完全匹配，包括分区列的位置。用select *可能因为源表和目标表的列顺序不一致，导致数据写入错位，最终计数异常。

解决方法：
明确写出所有列名，确保和目标表结构对齐：

# 假设目标表列顺序为：col1, col2, col3, data_date, data_product
fact = spark.sql(f"""
    insert overwrite table {usageWideFactTable} partition (data_date, data_product)
    select col1, col2, col3, data_date, data_product 
    from usage_fact_staging
""")

5. 调试验证步骤 #

每次运行前先对比临时视图和目标表的行数，快速定位问题环节：

# 查看临时视图总行数
print("临时视图行数：", spark.sql("select count(*) from usage_fact_staging").collect()[0][0])
# 查看目标表当前总行数
print("目标表当前行数：", spark.sql(f"select count(*) from {usageWideFactTable}").collect()[0][0])
# 执行insert后再查看目标表行数
fact = spark.sql(...)
print("插入后目标表行数：", spark.sql(f"select count(*) from {usageWideFactTable}").collect()[0][0])

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