<Spark><Advanced Programming>

Introduction

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

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.

Broadcast Variables

• Spark支持的另一种共享变量的方式：broadcast variables。允许程序高效地发送大的，只读的value给所有worker nodes。
• 你可能会想到说Spark会自动地发送闭包中所引用的变量到worker node。这是方便的，但同时也是很不高效的。因为：
  1. 缺省的task launching 机制是对small task sizes优化的；
  2. 你可能在多个并行操作中使用相同的变量，但是Spark会分别为每个Operation发送一次。考虑下面的操作：

# 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))) 

 　　　　上面的代码中，如果signPrefixes是一个很大的table，那么将该表从master传到每个slave将是很昂贵的。而且如果后期还要用到signPrefixes，它将会被再次发送到每个节点。通过将signPrefixes变成broadcast变量可以解决这个问题。如下：

　　　　

// 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")

•  总的来说，broadcast变量的使用分为以下几步：
  1. 创建一个Broadcast[T]的SparkContext.broadcast对象of type T。T可以是任何类型，只要它是Serializable的。
  2. 同value property访问该值
  3. 该变量只会被传送给每个node一次，而且被当做read-only来使用，也就是说更新不会propagated到其他nodes。(最简单的满足read-only方式的方法是broadcast a primitive value or a reference to an immutable object)。
• broadcast variable就是一个spark.broadcast.Broadcast[T] 类型的对象，它wraps一个类型T的值。我们可以在tasks中访问该值。该值只发送到每个节点一次，通过使用高效的BitTorrent-like communication mechanism。

Optimizing Broadcasts

• 当你broadcast a large values, it's important to choose a data serialization format that is both fast and compact。Scala缺省使用的JAVA Serialization库是很低效的(JAVA Serialization只对原生类型的数组高效)。因此你可以通过选择一个不同的序列化库(通过使用spark.serializer property)来优化。

Broadcast Internals

• 只能broadcast只读的变量：这涉及到一致性问题。
• Broadcast到节点而非task：因为每个task是一个线程，而且同一个进程允许的tasks都属于同一个application，因此只要在每个节点(executor)上放一份可以被task共享。
• 怎么实现Broadcast？
  1. 分发task的时候先分发bdata的元信息：Driver先建一个本地文件夹以存放需要Broadcast的data，并启动一个可以访问该文件夹的httpServer；
  2. 调用val bdata = sc.broadcast(data)时就把data写入文件夹，同时写入driver自己的blockManager中(StorageLevel为内存+磁盘)，获得一个blockID，类型为BroadcastBlockId；
  3. 调用rdd.transformation(func)时，若func用到了bdata，那么driver submitTask()时会将bdata和func一起进行序列化得到serialized task，注意序列化时不会序列化bdata包含的data，因为实际的data可能会很大。
  4. 在executor反序列化task的时候，同时会反序列化task中bdata对象，这时会调用bdata.readObject()，该方法会先去本地blockManager询问bdata的data是否在blockManager内，若不在就用HttpBroadcast/TorrentBroadcast去将data fetch过来。得到data后将其存在blockManager中，这样后续的task如果需要bdata就不需要再去fetch data了。
• 对比Hadoop中的DistributedCache：hadoop的这种需要先将文件上传到HDFS，这种方式的主要问题就是浪费资源，若某节点要允许来自同一job的多个mapper，那么公共数据会在该节点上保留多份(每个task的工作目录会有一份)。但是Hadoop这种方式的好处是单点瓶颈不明显，因为data在HDFS上分为多个block，而且有副本，这样只要所有的task不是同时去同一个节点fetch数据，网络拥塞就不会很严重。

Working on a Per-Partition Basis

•  Working with data on a per-partition basis allows us to avoid redoing setup work for each data item.
• 比如说打开一个数据库连接或创建一个随机数生成器都是setup steps。
• 为此，Spark提供了per-partition version of map and foreach，来提供让你为RDD的每个partition运行一次来减少这些操作。
• 【这种方式有点像MapReduce中每个Mapper/Reducer的setup()方法。】
• eg：如下的例子中，提供使用partition-based操作，来share一个连接池，并重用JSON parser

val contactsContactLists = validSigns.distonct().mapPartitions{
    sign  =>
    val mapper = createMapper()
    val client = new HttpClient()
    client.start()
    // create http request
    signs.map{ sign =>
        createExchangeForSign(sign)
    // fetch responses
    }.map{ case(sign, exchange) =>
        (sign, readExchangeCallLog(mapper, exchange))
    }.filter( x => x._2 != null) // remove empty CallLogs
}

• 对于per-partition的操作，Spark给予函数一个Iterator of the elements in that partition. 然后我们提供返回一个Iterator来返回值。一下是一些per-partition operators：

Function Name	We are called with	We return	Function signature on RDD[T]
mapPartitons()	Iterator of elements in that partiton	Iterator of our return elements	f: (Iterator[T]) -> Iterator[U]
mapPartitionsWithIndex()	Integer of partition number, and Iterator of the elements in that partition.	Iterator of our return elements	f: (Int, Iterator[T]) -> Iterator[U]
foreachPartition()	Iterator of the elements	Nothing	f: (Iterator[T]) -> Unit

• 除了避免setup work的开销，我们有时可以使用mapPartitions()来避免创建对象的开销。

FYI

• Broadcast Internal