Spark Core快速入门系列(12) | 变量与累加器问题

package Demo
import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

/**
 **
@author 不温卜火
 **
 * @create 2020-08-01 12:18
 **
 *         MyCSDN ：  https://buwenbuhuo.blog.csdn.net/
 *
 */
object AccDemo1 {
  def main(args: Array[String]): Unit = {
    val conf = new SparkConf().setAppName("Practice").setMaster("local[2]")
    val sc = new SparkContext(conf)
    val p1 = Person(10)
    // 将来会把对象序列化之后传递到每个节点上
    val rdd1 = sc.parallelize(Array(p1))
    val rdd2: RDD[Person] = rdd1.map(p => {p.age = 100; p})

    rdd2.count()
    // 仍然是 10
    println(p1.age)
  }

}
case class Person(var age:Int)

package Demo
import org.apache.spark.rdd.RDD
import org.apache.spark.util.LongAccumulator
import org.apache.spark.{SparkConf, SparkContext}

/**
 **
@author 不温卜火
 **
 * @create 2020-08-01 12:22
 **
 *         MyCSDN ：  https://buwenbuhuo.blog.csdn.net/
 *
 */
object AccDemo2 {
  def main(args: Array[String]): Unit = {
    val conf = new SparkConf().setAppName("Practice").setMaster("local[2]")
    val sc = new SparkContext(conf)
    val rdd: RDD[String] = sc.textFile("D:/words.txt")
    // 得到一个 Long 类型的累加器.  将从 0 开始累加
    val emptyLineCount: LongAccumulator = sc.longAccumulator
    rdd.foreach(s => if (s.trim.length == 0) emptyLineCount.add(1))
    println(emptyLineCount.value)
  }

}

package Demo
import java.util
import java.util.{ArrayList, Collections}

import org.apache.spark.util.AccumulatorV2

/**
 **
@author 不温卜火
 **
 * @create 2020-08-01 12:56
 **
 *         MyCSDN ：  https://buwenbuhuo.blog.csdn.net/
 *
 */
object MyAccDemo {
  def main(args: Array[String]): Unit = {

  }
}

class MyAcc extends AccumulatorV2[String, java.util.List[String]] {
  private val _list: java.util.List[String] = Collections.synchronizedList(new ArrayList[String]())
  override def isZero: Boolean = _list.isEmpty

  override def copy(): AccumulatorV2[String, util.List[String]] = {
    val newAcc = new MyAcc
    _list.synchronized {
      newAcc._list.addAll(_list)
    }
    newAcc
  }

  override def reset(): Unit = _list.clear()

  override def add(v: String): Unit = _list.add(v)

  override def merge(other: AccumulatorV2[String, util.List[String]]): Unit =other match {
    case o: MyAcc => _list.addAll(o.value)
    case _ => throw new UnsupportedOperationException(
      s"Cannot merge ${this.getClass.getName} with ${other.getClass.getName}")
  }

  override def value: util.List[String] = java.util.Collections.unmodifiableList(new util.ArrayList[String](_list))
}

package Demo

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}


/**
 **
 *
*@author 不温卜火
 **
 * @create 2020-08-01 12:57
 **
 *         MyCSDN ：  https://buwenbuhuo.blog.csdn.net/
 *
 */
object MyAccDemo1 {
  def main(args: Array[String]): Unit = {
    val pattern = """^\d+$"""
    val conf = new SparkConf().setAppName("Practice").setMaster("local[2]")
    val sc = new SparkContext(conf)
    // 统计出来非纯数字, 并计算纯数字元素的和
    val rdd1 = sc.parallelize(Array("abc", "a30b", "aaabb2", "60", "20"))

    val acc = new MyAcc
    sc.register(acc)
    val rdd2: RDD[Int] = rdd1.filter(x => {
      val flag: Boolean = x.matches(pattern)
      if (!flag) acc.add(x)
      flag
    }).map(_.toInt)
    println(rdd2.reduce(_ + _))
    println(acc.value)
  }

}

scala> val broadcastVar = sc.broadcast(Array(1, 2, 3))
broadcastVar: org.apache.spark.broadcast.Broadcast[Array[Int]] = Broadcast(0)

scala> broadcastVar.value
res0: Array[Int] = Array(1, 2, 3)