本地运行KeyValueGroupedDataset的flatMapGroups结果异常排查

这个问题是Spark 2.x早期版本的已知Bug，根源是flatMapGroups在处理不可变序列（如Seq）的迭代器时存在复用问题：当Spark内部重复遍历返回的序列迭代器时，由于Seq的迭代器是一次性的，遍历结束后不会重置，导致每次只能取到最后一个元素，最终输出全为3。而Databricks通常使用的是修复过该Bug的Spark版本（比如2.4.5及以上，或3.x系列），所以能得到正确结果。

解决方法 #

有两种可行的修复方式：

1. 升级Spark版本
   将本地环境的Spark依赖升级到2.4.5或更高版本，或者直接切换到Spark 3.x系列，该Bug在这些版本中已被官方修复。

2. 修改代码避免迭代器复用
   不直接返回Seq，而是返回每次调用都会生成新迭代器的类型，比如Iterator：

   import org.apache.spark.sql.SparkSession
   
   object FlatMapGroupsFix {
     def main(args: Array[String]): Unit = {
       val spark: SparkSession = SparkSession.builder
         .appName("Bug fix")
         .master("local")
         .getOrCreate()
       import spark.implicits._
       Seq(1, 2).toDS
         .groupByKey(x => x)
         .flatMapGroups((_, _) => Iterator(1, 2, 3)) // 改用Iterator
         .show
       spark.stop()
     }
   }
   

   改用Iterator能确保每次调用都生成全新的迭代器，避免复用导致的元素重复问题。

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