廣播變量和累加器

廣播變量

• 廣播變量理解圖

未使用廣播變量：?

package SparkRadioimport org.apache.spark.rdd.RDD import org.apache.spark.{SparkConf, SparkContext}object Demo01 {def main(args: Array[String]): Unit = {val conf = new SparkConf().setAppName("redio").setMaster("local")val sc = new SparkContext(conf)sc.setLogLevel("error")val list = List[String]("honey","lucy")//Distribute a local Scala collection to form an RDD.val nameList: RDD[String] = sc.parallelize(List[String]("george", "honey", "lucy"))val result = nameList.filter(name => {!list.contains(name)})result.foreach(println)//george} }

使用 廣播變量

package SparkRadioimport org.apache.spark.broadcast.Broadcast import org.apache.spark.rdd.RDD import org.apache.spark.{SparkConf, SparkContext}object Demo01 {def main(args: Array[String]): Unit = {val conf = new SparkConf().setAppName("redio").setMaster("local")val sc = new SparkContext(conf)sc.setLogLevel("error")val list = List[String]("honey","lucy")//廣播變量val bc: Broadcast[List[String]] = sc.broadcast(list)//Distribute a local Scala collection to form an RDD.val nameList: RDD[String] = sc.parallelize(List[String]("george", "honey", "lucy"))val result = nameList.filter(name => {val innerList: List[String] = bc.value!innerList.contains(name)})result.foreach(println)//george} }

注意事項

• 能不能將一個RDD使用廣播變量廣播出去？

不能，因為RDD是不存儲數據的。可以將RDD的結果廣播出去。

• 廣播變量只能在Driver端定義，不能在Executor端定義。
• 在Driver端可以修改廣播變量的值，在Executor端無法修改廣播變量的值。

廣播變量的意義

如果我們要在分布式計算里面分發大對象，例如：字典，集合，黑白名單等，這個都會由Driver端進行分發，一般來講，如果這個變量不是廣播變量，那么每個task就會分發一份，這在task數目十分多的情況下Driver的帶寬會成為系統的瓶頸，而且會大量消耗task服務器上的資源，如果將這個變量聲明為廣播變量，那么只是每個executor擁有一份，這個executor啟動的task會共享這個變量，節省了通信的成本和服務器的資源。

?

---

---

累加器

• 累加器理解圖

?

package SparkRadioimport org.apache.spark.sql.SparkSessionobject AccumulatorTest {def main(args: Array[String]): Unit = {//另一種方式而已val spark: SparkSession = SparkSession.builder().appName("test").master("local").getOrCreate()val sc = spark.sparkContextsc.setLogLevel("error")val rddd1 = sc.textFile("./data/words.txt")var i = 0val rdd2 = rddd1.map(one => {i += 1println(s"Executor i = $i")one})rdd2.collect()println(s"i = $i")} } package SparkRadioimport org.apache.spark.sql.SparkSessionobject AccumulatorTest {def main(args: Array[String]): Unit = {//另一種方式而已val spark: SparkSession = SparkSession.builder().appName("test").master("local").getOrCreate()val sc = spark.sparkContextsc.setLogLevel("error")val accumulator = sc.longAccumulatorval rddd1 = sc.textFile("./data/words.txt")val rdd2 = rddd1.map(one => {accumulator.add(1)one})rdd2.collect()println(s"accumulator = ${accumulator.value}")//accumulator = 5} }

?

注意事項

• 累加器在Driver端定義賦初始值，累加器只能在Driver端讀取，在Excutor端更新。

累加器的意義

在spark應用程序中，我們經常會有這樣的需求，如異常監控，調試，記錄符合某特性的數據的數目，這種需求都需要用到計數器，如果一個變量不被聲明為一個累加器，那么它將在被改變時不會再driver端進行全局匯總，即在分布式運行時每個task運行的只是原始變量的一個副本，并不能改變原始變量的值，但是當這個變量被聲明為累加器后，該變量就會有分布式計數的功能。

?

總結

以上是生活随笔為你收集整理的Spark _15 _广播变量和累加器的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。