Spark 累加器

defgetMap(spark:SparkSession,data:DataFrame){
//通过collectionAccumulator构造Map关系
valmyAccumulator=spark.sparkContext.collectionAccumulator[(String,Long)]
data.foreach(
row=>{
valname=row.getAs[String]("name")
valage=row.getAs[Long]("age")
myAccumulator.add(name,age)
}
)
valaiterator:util.Iterator[(String,Long)]=myAccumulator.value.iterator()
varnewMap:Map[String,Long]=Map()
while(aiterator.hasNext){
vala=aiterator.next()
valkey=a._1
valvalue=a._2
if(!newMap.contains(key)){
newMap+=(key->value)
}
else{
valoldvalue=newMap(key)
newMap+=(key->(oldvalue+value))
}
}
}