怎么解析SPARK foreach循环中的变量问题,针对这个问题,这篇文章详细介绍了相对应的分析和解答,希望可以帮助更多想解决这个问题的小伙伴找到更简单易行的方法。
在spark算子中引用的外部变量,其实是变量的副本,在算子中对其值进行修改,只是改变副本的值,外部的变量还是没有变。
通俗易懂的讲就是foreach里的变量带不出来的,除非用map,将结果作为rdd返回
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
}
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())
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
关于怎么解析SPARK foreach循环中的变量问题问题的解答就分享到这里了,希望以上内容可以对大家有一定的帮助,如果你还有很多疑惑没有解开,可以关注亿速云行业资讯频道了解更多相关知识。
亿速云「云服务器」,即开即用、新一代英特尔至强铂金CPU、三副本存储NVMe SSD云盘,价格低至29元/月。点击查看>>
免责声明:本站发布的内容(图片、视频和文字)以原创、转载和分享为主,文章观点不代表本网站立场,如果涉及侵权请联系站长邮箱:is@yisu.com进行举报,并提供相关证据,一经查实,将立刻删除涉嫌侵权内容。
原文链接:https://my.oschina.net/u/3209854/blog/4989655