Introduction

• 介绍两种共享变量的方式：
  • accumulators：聚集信息
  • broadcast variables：高效地分布large values
• 介绍对高setup costs任务的批操作，比如查询数据库时连接数据的消耗。

Accumulators

• 当我们向Spark传送函数时(比如map()函数或给filter()的condition)，他们可以使用driver program中在他们定义之外的变量。但是cluster中的每个task都get a new copy of each variable，并且更新那些副本而不会传播到driver。
• Spark的共享变量--accumulators和broadcast variables，通过两种通信模式（聚集结果和广播）放松了这种限制。
• Accumulators提供了一种简单的语法，用于从worker聚集values返回到driver program。
• Accumulators的一个最广泛的用例就是统计在job执行期间发生的events数，用于debugging。

• val sc = new SparkContext(...)
  val file = sc.textFile("file.txt")
  
  val blankLines = sc.accumulator(0)    // create an Accumulator[Int] initialized to 0
  
  val callSigns = file.flatMap(line => { 
      if (line == ""){
          blankLines += 1   // Add to the accumulator
      }
      line.split(" ")
  })
  callSigns.saveAsTextFile("output.txt")
  println("Blank lines: " + blankLines.value)

  注意只有在调用saveAsTextFile之后才能得到正确的count of blankLines，因为transfomation是lazy的。

• Summary of accumulators:
  • 我们提供调用SparkContext.accumulator(initialValue)方法创建一个accumulator，它会返回一个org.apache.spark.Accumulator[T]对象，T是initialValue的类型；
  • Spark闭包中的worker code可以通过 += 来add accumulator；
  • driver program可以call accumulator的value property来访问accumulator的值。
• 注意worker node不可以访问accumulator的value --> 因为从tasks的角度来看，accumulators是write-only的变量。这样的设定使得accumulators可以被高效地实现，而不需要每次更新的时候都通信。

Accumulators and Fault Tolerance

• Spark通过re-executikng failed or slow tasks来处理failed or slow machines。
• 那么错误与Accumulators之间呢？ --> 对于在actions中使用的Accumulators，Spark仅仅对每个Accumulator执行一次每个task的更新。因此，如果我们想要一个绝对可靠的value counter，而不用考虑failures或者多次赋值，那么我们必须将操作放到类似foreach()这样的操作中。
• 对于在transformation中使用的Accumulators，这种保证是不存在的。在transformation中对Accumulators的更新可能执行多次。所以对transformation中的Accumulators最好只在debug时用。[for version1.2.0]

Custom Accumulators

• Spark支持Double、Long、Float、Int等类型的Accumulators。同时还提供了一个自定义Accumulators type的API，来实现自定义Accumulators types以及支持自定义聚集操作(比如找最大值而不是add)。
• 可支持的操作必须是commutative and associative的。[感觉就像是顺序对操作不重要就可以了]
  • An operation op is commutative if a op b = b op a for all values a, b. 

  • An operation op is associative if (a op b) op c = a op (b op c) for all values a, b, and c.

# Look up the locations of the call signs on the # RDD contactCounts. We load a list of call sign # prefixes to country code to support this lookup. 
signPrefixes = loadCallSignTable()

def processSignCount(sign_count, signPrefixes):
    country = lookupCountry(sign_count[0], signPrefixes) count = sign_count[1]
    return (country, count)

countryContactCounts = (contactCounts
                       .map(processSignCount)
                       .reduceByKey((lambda x, y: x+ y))) 

// Look up the countries for each call sign for the
// contactCounts RDD. We load an array of call sign
// prefixes to country code to support this lookup.
val signPrefixes = sc.broadcast(loadCallSignTable())

val countryContactCounts = contactCounts.map{case (sign, count) =>
    val country = lookupInArray(sign, signPrefixes.value)
    (country, count)
}.reduceByKey((x, y) => x + y) 

countryContactCounts.saveAsTextFile(outputDir + "/countries.txt")

<Spark><Advanced Programming>