spark 共享变量

关于累计器， 广播变量， 参考：
http://blog.csdn.net/u013468917/article/details/70617085(累加器主要参考了这篇文章)
https://www.cnblogs.com/liuliliuli2017/p/6782687.html(广播变量)
http://blog.csdn.net/leen0304/article/details/78866353

对于多节点变量的共享， 我们可以依赖Redis混存数据库来实现。
Spark已经提供了两种特定的共享变量，来完成节点间变量的共享： 累加器和广播变量

累加器accumulator

继承于抽象类AccumulatorV2

// 输入两个类型参数 IN, OUT
abstract class AccumulatorV2[IN, OUT] extends Serializable {
  private[spark] var metadata: AccumulatorMetadata = _
  private[this] var atDriverSide = true

 //  注册累计器， 生成一个ID 和 Name
// countFailedValues对于失败的task是否计数， 默认false
 private[spark] def register(
      sc: SparkContext,
      name: Option[String] = None,
      countFailedValues: Boolean = false): Unit = {
    if (this.metadata != null) {
      throw new IllegalStateException("Cannot register an Accumulator twice.")
    }
    this.metadata = AccumulatorMetadata(AccumulatorContext.newId(), name, countFailedValues)
    AccumulatorContext.register(this)
    sc.cleaner.foreach(_.registerAccumulatorForCleanup(this))
  }
// 每个累计器只能和注册一次， 使用之前必须注册
final def isRegistered: Boolean =
    metadata != null && AccumulatorContext.get(metadata.id).isDefined

  private def assertMetadataNotNull(): Unit = {
    if (metadata == null) {
      throw new IllegalStateException("The metadata of this accumulator has not been assigned yet.")
    }
  }

final def id: Long = {
    assertMetadataNotNull()
    metadata.id
}

/**
   * Returns the name of this accumulator, can only be called after registration.
   */
 final def name: Option[String] = {
    assertMetadataNotNull()
    if (atDriverSide) {
metadata.name.orElse(AccumulatorContext.get(id).flatMap(_.metadata.name))
    } else {
      metadata.name
    }
  }

在继承类中必须需要提供实现的方法：
isZeros(): 累计器空的判断, 返回Boolean
copy(): 拷贝实现， 返回CollectionAccumulator
reset()： 重置
add()：累计
merge(): 合并不同节点的累计器会使用
value： 累计器实际容器

/**
   * Returns if this accumulator is zero value or not. e.g. for a counter accumulator, 0 is zero
   * value; for a list accumulator, Nil is zero value.
   */
  def isZero: Boolean

/**
   * Creates a new copy of this accumulator.
   */
  def copy(): AccumulatorV2[IN, OUT]

 /** 清空
   * Resets this accumulator, which is zero value. i.e. call `isZero` must
   * return true.
   */
  def reset(): Unit

/** 累加的具体实现
   * Takes the inputs and accumulates.
   */
  def add(v: IN): Unit

/** 合并累计器到当前累计器
   * Merges another same-type accumulator into this one and update its state, i.e. this should be merge-in-place.
   */
  def merge(other: AccumulatorV2[IN, OUT]): Unit

/** 调用value的返回
   * Defines the current value of this accumulator
   */
  def value: OUT

Spark在SparkContext类中的提供了longAccumulator和doubleAccumulator， collectionAccumulator的累加器, 各有一个带name和不带name的重载接口, 具体实现代码在org.apache.spark.util

def collectionAccumulator[T]: CollectionAccumulator[T] = {
    val acc = new CollectionAccumulator[T]  // 定义累计器
    register(acc)  // 注册， name可选
    acc
  }

  /**
   * Create and register a `CollectionAccumulator`, which starts with empty list and accumulates
   * inputs by adding them into the list.
   */
  def collectionAccumulator[T](name: String): CollectionAccumulator[T] = {
    val acc = new CollectionAccumulator[T]
    register(acc, name)
    acc
  }

CollectionAccumulator 定义了一个list类型的累加器

class CollectionAccumulator[T] extends AccumulatorV2[T, java.util.List[T]] {
// 中间变量
  private val _list: java.util.List[T] = Collections.synchronizedList(new ArrayList[T]())
// 实现 isZero
  override def isZero: Boolean = _list.isEmpty
// 实现 copyAndReset， AccumulatorV2中提供了方法： 先copy, 再reset， 非必须
  override def copyAndReset(): CollectionAccumulator[T] = new CollectionAccumulator
// 实现 copy(), 返回CollectionAccumulator
  override def copy(): CollectionAccumulator[T] = {
    val newAcc = new CollectionAccumulator[T]
    _list.synchronized {
      newAcc._list.addAll(_list)
    }
    newAcc
  }
// 实现reset(), 原位操作
  override def reset(): Unit = _list.clear()
// 实现add()， 或者java list可以改为 mutable.List
  override def add(v: T): Unit = _list.add(v)
// 合并， match 模式匹配很好用，一堆try catch的作用, addAll
  override def merge(other: AccumulatorV2[T, java.util.List[T]]): Unit = other match {
    case o: CollectionAccumulator[T] => _list.addAll(o.value)
    case _ => throw new UnsupportedOperationException(
      s"Cannot merge ${this.getClass.getName} with ${other.getClass.getName}")
  }
// value的方法实现
  override def value: java.util.List[T] = _list.synchronized {
    java.util.Collections.unmodifiableList(new ArrayList[T](_list))
  }
// 新增方法， clear后设置新的list
  private[spark] def setValue(newValue: java.util.List[T]): Unit = {
    _list.clear()
    _list.addAll(newValue)
  }
}

自定义实现Set的累加器

class  LogAccumulator extends AccumulatorV2[String,java.util.Set[String]]{
  private val _logArray: java.util.Set[String] = new java.util.HashSet[String]()

  override def isZero: Boolean = {
    _logArray.isEmpty
  }

  override def reset(): Unit = {
    _logArray.clear()
  }

  override def add(v: String): Unit = {
    _logArray.add(v)
  }

  override def merge(other: AccumulatorV2[String, java.util.Set[String]]): Unit = {
    other match {
      case o: LogAccumulator => _logArray.addAll(o.value)
    }
  }

  override def value: java.util.Set[String] = {
    java.util.Collections.unmodifiableSet(_logArray)
  }

  override def copy(): AccumulatorV2[String, util.Set[String]] = {
    val newAcc = new LogAccumulator()
    _logArray.synchronized{
      newAcc._logArray.addAll(_logArray)
    }
    newAcc
  }
}

注：
1， java.util.Collections.unmodifiableSet 参考 https://www.yiibai.com/java/util/java_util_collections.html 此方法返回指定列表的不可修改视图。

累加器的坑：

1， spark RDD lazy操作， 可能导致累加器多加或少加，参照：http://blog.csdn.net/u013468917/article/details/70617085
所以使用完add累计器记得cache()