| 下面这个map在循环里面有值，到循环外的map里就没有值了。。。
object Test {
def main(args: Array[String]): Unit = {
val conf = new SparkConf();
conf.setMaster("local[1]");
conf.setAppName("WcAppTask");
val sc = new SparkContext(conf);
val fileRdd = sc.parallelize(Array(("imsi1","2018-07-29 11:22:23","zd-A"),("imsi2","2018-07-29 11:22:24","zd-A"),("imsi3","2018-07-29 11:22:25","zd-A")))
val result = mutable.Map.empty[String,String]
fileRdd.foreach(input=>{
val str = (input._1+"/t"+input._2+"/t"+input._3).toString;
result += (input._1.toString -> str)
println(result.size) //返回1，2,3
})
println(result.size) //返回0
}
} |
运行的结果
1
2
3
0
我们需要的结果当然是1,2,3,3
寻找原因：
在spark算子中引用的外部变量，其实是变量的副本，在算子中对其值进行修改，只是改变副本的值，外部的变量还是没有变。
通俗易懂的讲就是foreach里的变量带不出来的，除非用map，将结果作为rdd返回
解决方案：1使用广播变量
object foreachtest {
def main(args: Array[String]): Unit = {
val conf = new SparkConf()
conf.setMaster("local[1]")
conf.setAppName("WcAppTask")
val sc = new SparkContext(conf)
sc.setLogLevel("WARN")
val fileRdd = sc.parallelize(Array(("imsi1","2018-07-29 11:22:23","zd-A"),("imsi2","2018-07-29 11:22:24","zd-A"),("imsi3","2018-07-29 11:22:25","zd-A")))
val result = mutable.Map.empty[String,String]
val resultBroadCast: Broadcast[mutable.Map[String, String]] =sc.broadcast(result)
fileRdd.foreach(input=>{
val str = (input._1+"/t"+input._2+"/t"+input._3).toString
resultBroadCast.value += (input._1.toString -> str)
println(resultBroadCast.value.size) //返回1,2.3
})
println(result.size) //返回3
}
运行的结果
1
2
3
3
解决方案2：使用累加器
val accum = sc.collectionAccumulator[mutable.Map[String, String]]("My Accumulator")
fileRdd.foreach(input => {
val str = input._1 + "/t" + input._2 + "/t" + input._3
accum.add(mutable.Map(input._1 -> str))
})
println(accum.value.size())
运行结果 3
方案3：和2类似
val longaa= sc.longAccumulator("count")
fileRdd.foreach(input=>{
val str = (input._1+"/t"+input._2+"/t"+input._3).toString
longaa.add(1L)
})
println(longaa.count) //返回3
运行结果 3
注意事项：
累加器的使用陷阱（未完待续）
| 
转播
