六十四、Spark-分別統計各個單詞個數及特殊字符總個數

共享變數

廣播變數（Broadcast Variables）：廣播變數用來把變數在所有節點的記憶體之間進行共享，在每個機器上快取一個只讀的變數，而不是為機器上的每個任務都生成一個副本，簡單理解：減少記憶體，減小計算壓力;

累加器（Accumulators）：累加器支持在所有不同節點之間進行累加計算(比如計數或者求和);

需求說明：以詞頻統計WordCount程式為例，處理特殊資料，包括非單詞符號，做WordCount的同時統計出特殊字符的數量

目錄

共享變數

原資料展示

業務邏輯

完整代碼

程式運行

專案總結

原資料展示

注：原資料雜亂無章，與所需單詞混淆，間隔且不等

業務邏輯

1、創建本地環境，并設定日志提示級別

val conf: SparkConf = new SparkConf().setAppName("spark").setMaster("local[*]")
val sc: SparkContext = new SparkContext(conf)
sc.setLogLevel("WARN")

2、創建累加器

val mycounter: LongAccumulator = sc.longAccumulator("mycounter")

3、定義特殊字符集合

val ruleList: List[String] = List(",", ".", "!", "#", "$", "`", "~", "@", "?", "*", "^")

4、將集合作為廣播變數廣播到各個節點

val broadcast: Broadcast[List[String]] = sc.broadcast(ruleList)

5、加載資料，創建RDD

val lines: RDD[String] = sc.textFile("data/input/words2.txt")

6、過濾篩選

val wordcountResult: RDD[(String, Int)] = lines.filter(StringUtils.isNoneBlank(_))
      .flatMap(_.split("\\s+"))
      .filter(ch => {
        //獲取廣播資料
        val list: List[String] = broadcast.value
        if (list.contains(ch)) { //特殊字符
          mycounter.add(1)
          false
        } else { //單詞
          true
        }
      }).map((_, 1))
      .reduceByKey(_ + _)

7、輸出單詞統計及特殊字符

wordcountResult.foreach(println)
val chResult: lang.Long = mycounter.value
println("特殊字符的數量:" + chResult)

完整代碼

package org.example.spark

import java.lang
import org.apache.commons.lang3.StringUtils
import org.apache.spark.broadcast.Broadcast
import org.apache.spark.rdd.RDD
import org.apache.spark.util.LongAccumulator
import org.apache.spark.{SparkConf, SparkContext}

/**
 * Author tuomasi
 * Desc 演示RDD的共享變數
 */
object RDD_ShareVariable {
  def main(args: Array[String]): Unit = {
    //TODO 0.env/創建環境
    val conf: SparkConf = new SparkConf().setAppName("spark").setMaster("local[*]")
    val sc: SparkContext = new SparkContext(conf)
    sc.setLogLevel("WARN")

    //創建計數器/累加器
    val mycounter: LongAccumulator = sc.longAccumulator("mycounter")
    //定義一個特殊字符集合
    val ruleList: List[String] = List(",", ".", "!", "#", "$", "`", "~", "@", "?", "*", "^")
    //將集合作為廣播變數廣播到各個節點
    val broadcast: Broadcast[List[String]] = sc.broadcast(ruleList)

    //TODO 1.source/加載資料/創建RDD
    val lines: RDD[String] = sc.textFile("data/input/words2.txt")

    //TODO 2.transformation
    val wordcountResult: RDD[(String, Int)] = lines.filter(StringUtils.isNoneBlank(_))
      .flatMap(_.split("\\s+"))
      .filter(ch => {
        //獲取廣播資料
        val list: List[String] = broadcast.value
        if (list.contains(ch)) { //特殊字符
          mycounter.add(1)
          false
        } else { //單詞
          true
        }
      }).map((_, 1))
      .reduceByKey(_ + _)
    
    //TODO 3.sink/輸出
    wordcountResult.foreach(println)
    val chResult: lang.Long = mycounter.value
    println("特殊字符的數量:" + chResult)
  }
}

程式運行

原資料：

控制臺列印：

注：通過對比，該程式實作了單詞與特殊字符的分別統計

專案總結

使用廣播變數的好處：

1、Driver每次分發任務的時候會把task和計算邏輯的變數發送給Executor，不使用廣播變數，在每個Executor中有多少個task就有多少個Driver端變數副本，這樣會導致消耗大量的記憶體導致嚴重的后果，
2、使用廣播變數的好處，不需要每個task帶上一份變數副本，而是變成每個節點的executor才一份副本，這樣的話， 就可以讓變數產生的副本大大減少；