Spark多操作致失败：DataFrame写入Hive后Count广播超时排查

一、先搞定窗口操作无分区的警告（核心根源） #

你看到的No Partition Defined for Window operation!警告，说明你的DataFrame或者output_cols的计算逻辑里，存在未指定分区键的窗口函数操作。这种情况Spark会把所有数据shuffle到单个分区，不仅拖慢性能，还会引发后续操作的广播/Shuffle超时。

解决办法：

• 检查上游生成df的逻辑，或者output_cols里有没有窗口函数（比如row_number()、rank()这类）。
• 给窗口函数添加partitionBy指定合理的分区键，比如和Hive表的分区字段一致，或者选择高基数字段。示例：

  // 错误写法：无分区的窗口
  val window = Window.orderBy("col1")
  // 正确写法：指定分区键
  val window = Window.partitionBy("partition_col1").orderBy("col1")
  

• 如果确实不需要分区（不推荐），可以调大spark.sql.windowExec.buffer.spill.threshold参数临时规避，但优先还是添加分区键。

二、解决写入后DF缓存丢失+广播超时的问题 #

1. 缓存消失的原因
   Spark的缓存会按LRU策略自动清理，执行insertInto这类IO密集型操作时，若集群资源紧张，之前persist的DF很可能被清理。而且你写入后再次调用df = df.persist()完全多余——如果缓存还在，重复persist无意义；如果缓存已被清理，重新persist会触发DF重计算，此时窗口操作遗留的单分区问题就会引发广播超时。

2. 具体解决步骤

• 取消重复persist：写入前已经执行过persist()和count()，DF已被缓存，写入后直接执行df.count()即可。优先解决窗口分区问题，否则重计算仍会踩坑。
• 更换缓存级别：若要确保缓存不轻易被清理，不要用默认的MEMORY_ONLY，改用persist(StorageLevel.MEMORY_AND_DISK_SER)（序列化存储，节省空间且更稳定）。示例：

  df = df.persist(StorageLevel.MEMORY_AND_DISK_SER)
  

• 排查广播变量来源：广播超时通常是因为大表被标记为广播表（默认阈值10MB，可通过spark.sql.autoBroadcastJoinThreshold调整）。检查DF是否包含大表join操作，要么调大广播阈值，要么给小表加/*+ BROADCASTJOIN(small_table) */提示，让大表走shuffle join。
• 验证缓存状态：写入前后可以用df.storageLevel查看缓存级别，用spark.sparkContext.getPersistentRDDs查看所有持久化的RDD，确认DF是否真的被缓存。

三、优化Hive写入的配置 #

你当前使用的写入配置中，部分参数在Spark 2.4.8+CDP环境下可调整得更合理：

• 去掉spark.sql.parquet.output.committer.class和spark.sql.sources.commitProtocolClass的强制指定，CDP默认的提交器已适配Hive，硬指定反而可能引发兼容性问题。
• 确保spark.sql.sources.partitionOverwriteMode设为dynamic时，目标Hive表的分区字段与DF中的字段完全匹配，避免多余的分区处理逻辑。

四、调试小技巧 #

• 打开Spark UI的Jobs页面，找到写入后count操作对应的Job，查看Stage中的Shuffle和广播情况，定位超时阶段。
• 查看SQL页面（DataFrame操作可转成SQL查看执行计划），分析是否存在意外的广播或单分区操作。
• 执行df.explain(true)查看DF的详细执行计划，确认窗口操作的分区情况及写入操作后的依赖链。

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