環(huán)境：
ubuntu16.04 64
偽分布式
使用的spark是2.3.1
scala 2.11.8
參考連接：
https://blog.csdn.net/android_xue/article/details/79780463#commentsedit

注意，這篇博客是對(duì)上述參考鏈接的總結(jié)和概括．

一句話講明，累加器干嘛的？
統(tǒng)計(jì)slava機(jī)中的數(shù)據(jù)的總數(shù)量的．
上代碼，
1.不使用累加器的代碼AccumulatorTest.scala：

import scala.collection.Map import scala.collection.mutable.ArrayBuffer import scala.util.Random import org.apache.spark.broadcast.Broadcast import org.apache.spark.ml.recommendation.{ALS, ALSModel} import org.apache.spark.sql.{DataFrame, Dataset, SparkSession} import org.apache.spark.sql.functions._ import org.apache.log4j.Logger import org.apache.log4j.Level import org.apache.spark.sql.execution.datasources.textobject AccumulatorTest {def main(args:Array[String]):Unit={Logger.getLogger("org").setLevel(Level.OFF)Logger.getLogger("akka").setLevel(Level.OFF)Logger.getRootLogger().setLevel(Level.ERROR) //這里是用來(lái)抑制一大堆log信息的. val spark = SparkSession.builder.appName("Intro").config("spark.master", "local").getOrCreate();spark.sparkContext.setLogLevel("ERROR")val sc = spark.sparkContextval linesRDD=sc.textFile("hdfs://master:9000/test/word.txt")var i=0;val result=linesRDD.map(s=>{i+=1})result.collect();println("word lines is:"+i)sc.stop()}}

1.啟動(dòng)Hadoop的HDFS系統(tǒng)
2.hdfs dfs -put word.txt hdfs://master:9000/test/
3.scalac AccumulatorTest1.scala
4.scala AccumulatorTest

2.使用累加器的代碼AccumulatorTest2.scala：

import scala.collection.Map import scala.collection.mutable.ArrayBuffer import scala.util.Random import org.apache.spark.broadcast.Broadcast import org.apache.spark.ml.recommendation.{ALS, ALSModel} import org.apache.spark.sql.{DataFrame, Dataset, SparkSession} import org.apache.spark.sql.functions._ import org.apache.log4j.Logger import org.apache.log4j.Level import org.apache.spark.sql.execution.datasources.textobject AccumulatorTest {def main(args: Array[String]): Unit = {Logger.getLogger("org").setLevel(Level.OFF)Logger.getLogger("akka").setLevel(Level.OFF)Logger.getRootLogger().setLevel(Level.ERROR) //這里是用來(lái)抑制一大堆log信息的. val spark = SparkSession.builder.appName("Intro").config("spark.master", "local").getOrCreate();val sc = spark.sparkContextsc.setLogLevel("ERROR")val linesRDD=sc.textFile("hdfs://master:9000/test/word.txt")val accumulator=sc.accumulator(0); //創(chuàng)建accumulator并初始化為0val result=linesRDD.map(s=> {accumulator.add(1)//有一條數(shù)據(jù)就增加1s//返回ｓ給result，意思也就是，把word.txt中的內(nèi)容賦值(可能會(huì)打亂順序)給ｒｅｓｕｌｔ})result.collect().foreach(println);println("words lines is :"+accumulator.value)sc.stop()} }

1.啟動(dòng)Hadoop的HDFS系統(tǒng)
2.hdfs dfs -put word.txt hdfs://master:9000/test/
3.scalac AccumulatorTest2.scala
4.scala AccumulatorTest

總結(jié)

以上是生活随笔為你收集整理的spark中累加器的使用（转）的全部?jī)?nèi)容，希望文章能夠幫你解決所遇到的問(wèn)題。

如果覺(jué)得生活随笔網(wǎng)站內(nèi)容還不錯(cuò)，歡迎將生活随笔推薦給好友。