当我在本地运行时,我看到了正确的记录输出。但是,当我在集群上运行时,输出是不同的,并且看起来不一致。甚至某些 mappedGroup 输出也是正确的。这是火花闭包的问题吗?不知道如何最好地描述我所看到的。
我可能不理解 mapGroups 并且并非每个组的所有值都进入 recordList 变量。
case class MyCaseClass (keyValue: int,c2: String,c3: String,c4: Double)
case class NewClass (thing1:String,thing2:String,thing3:String,thing4:String)
case class WorkTodo(myClassRecords: Seq[MyCaseClass]){
def toNewRecords: Seq[NewClass] = {
//e.g. work that requires all MyCaseClass.keyValue=1 to be in the list.
//This function would create new Java Objects to perform calculations and eventually output a set of NewClass records
}
val processedRecords = ds.as[MyCaseClass].groupByKey(_.keyValue)
.mapGroups {
case (v, iter) => {
var recordList = new ListBuffer[MyCaseClass]
iter.foreach {x=>
recordList += MyCaseClass(x.keyValue,x.c2,x.c3,x.c4)
}
WorkToDo(recordList).toNewRecords
}
}
PS 欢迎任何其他仍然使用数据集的解决方案:)