LearningSpark6.1

spark源码阅读环境

本章引入两类共享变量：累加器和广播变量。累加器聚合信息，广播变量有效的分布大值。引进批处理人南无，如数据库查询。覆盖和外部程序交互，如R脚本。

使用pipe()方法获得R的库

累加器

当我们传递如map()函数或条件给filter()，在驱动程序里，它们可以使用它们外部定义的变量，但是集群上运行的每个任务都获得每个变量的新的复制，这些复制的更新并没有传回驱动程序。spark共享变量：累加器和广播变量使用两种交互模式解决了这个限制：结果聚合和广播。

累加器提供了从worker聚合值到驱动程序的简单解析器.累加器的一个最常用的用法是任务执行中事件发生的计数。

例如：文件空行计数

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)

累加器总结

驱动程序调用

SparkContext.accumulator(initialValue)

方法创建一个累加器，返回类型是

org.apache.spark.Accumulator[T]

,T是initialValue的类型

worker 代码可以使用+=方法添加到累加器

驱动程序可以调用accumulator的value()方法获得值

worker节点无法获得累加器的值，累加器是只写变量。

累加器和容错

spark通过重复执行失败或慢的任务自动处理错误或者慢机器

用户累加器

spark提供int,double,Long,Float 类型的累加器，也支持用户自定义累加器和用户聚合操作，例如找到最大值而不是累加他们。用户累加器应该继承

AccumulatorParam

，参考

除了数值添加，我们可以使用任何满足交换律和结合律的操作，例如，我们可以跟踪最大值而不是累加他们。

广播变量

广播变量允许程序员缓存一个只读的变量在每台机器上面,而不是每个任务保存一份拷贝。例如,利用广播变量,我们能够

以一种更有效率的方式将一个大数据量输入集合的副本分配给每个节点。Spark也尝试着利用有效的广播算法去

分配广播变量,以减少通信的成本。

一个广播变量可以通过调用 SparkContext.broadcast(v) 方法从一个初始变量v中创建。广播变量是v的一个包装变量,它的值

可以通过 value 方法访问,下面的代码说明了这个过程:

scala>  val broadcastVar    =   sc.broadcast(Array(1,   2,  3))
broadcastVar:   spark.Broadcast[Array[Int]] =   spark.Broadcast(b5c40191-a864-4c7d-b9bf-d87e1a4e787c)
scala>  broadcastVar.value
res0:   Array[Int]  =   Array(1,    2,  3)

广播变量创建以后,我们就能够在集群的任何函数中使用它来代替变量v,这样我们就不需要再次传递变量v到每个节点上。

另外,为了保证所有的节点得到广播变量具有相同的值,对象v不能在广播之后被修改。

广播变量允许程序有效的发送大数据，只读值到每个节点供多种spark操作使用,例如发送查表结果或者机器学习算法中的大特征向量。

Working on a Per-Partition Basis

在每个分区处理数据让我们不用对每个数据项重复设置工作。例如打开数据库连接，创建随机数生成器都是我们希望对每个数据项避免的设置操作。

通过使用基于分区的操作，我们可以分享连接这个数据库的连接池避免设置多个连接。

下面的程序给我们一个输入RDD的每个分区中的元素迭代器，期望返回我们的结果的迭代器。

Example 6-11. Shared connection pool and JSON parser in Scala
val contactsContactLists = validSigns.distinct().mapPartitions{
signs =>
val mapper = createMapper()
val client = new HttpClient()
client.start()
// create http request
Working on a Per-Partition Basis
|
107signs.map {sign =>
createExchangeForSign(sign)
// fetch responses
}.map{ case (sign, exchange) =>
(sign, readExchangeCallLog(mapper, exchange))
}.filter(x => x._2 != null) // Remove empty CallLogs
}

除了避免设置工作，我们有时使用

mapPartitions()

避免对象创建。有时我们需要创建一个对象聚合不同类型的结果，

piping to 外部程序

R脚本

#!/usr/bin/env Rscript
library("Imap")
f <- file("stdin")
open(f)
while(length(line <- readLines(f,n=1)) > 0) {
# process line
contents <- Map(as.numeric, strsplit(line, ","))
mydist <- gdist(contents[[1]][1], contents[[1]][2],
contents[[1]][3], contents[[1]][4],
units="m", a=6378137.0, b=6356752.3142, verbose = FALSE)
write(mydist, stdout())
}

我们得到一种把stdin的每一行转换到stdout上的output的方法。现在我们需要让每个节点都能获得finddistance.R脚本，并且用脚本转换RDD：

// Compute the distance of each call using an external R program
// adds our script to a list of files for each node to download with this job
val distScript = "./src/R/finddistance.R"
val distScriptName = "finddistance.R"
sc.addFile(distScript)
val distances = contactsContactLists.values.flatMap(x => x.map(y =>
s"$y.contactlay,$y.contactlong,$y.mylat,$y.mylong")).pipe(Seq(
SparkFiles.get(distScriptName)))
println(distances.collect().toList)