Spark共享变量是指在Spark应用程序中,多个任务之间共享的变量。它可以被所有工作节点上的任务所使用,而不需要在每个任务中重新分发一份副本。Spark支持两种共享变量:广播变量和累加器。
广播变量是将数据复制到所有节点上的一种机制,它能够减少数据传输和存储开销,这样就能够减少计算时间。使用广播变量时,首先需要将数据通过broadcast()方法传递到所有工作节点上,然后在工作节点上使用value()方法来读取该数据。例如:
val broadcastVar = sc.broadcast(Array(1, 2, 3)) val result = rdd.map(x => x + broadcastVar.value.sum)
累加器是一个特殊的只写变量,它能够在集群中的所有Executor上进行共享读写。它能够帮助我们对一些共享的数值进行原子性的读写(即各个Executor之间不会相互影响)。使用时需要通过accumulator()方法来创建一个Accumulator对象,然后在Executor中使用add()方法来对Accumulator对象进行修改。例如:
val accum = sc.accumulator(0) rdd.foreach(x => accum += x) println("accum: " + accum.value)
一般情况下,当一个传递给Spark操作(例如map和reduce)的函数在远程节点上面运行时,Spark操作实际上操作的是这个函数所用变量的一个独立副本。这些变量被复制到每台机器上,并且这些变量在远程机器上的所有更新都不会传递回驱动程序。通常跨任务的读写变量是低效的,但是,Spark还是为两种常见的使用模式提供了两种有限的共享变量:广播变量(broadcast variable)和累加器(accumulator)
广播变量允许程序员缓存一个只读的变量在每台机器上面,而不是每个任务保存一份拷贝。例如,利用广播变量,我们能够以一种更有效率的方式将一个大数据量输入集合的副本分配给每个节点。(Broadcast variables allow theprogrammer to keep a read-only variable cached on each machine rather than shipping a copy of it with tasks.They can be used, for example,to give every node a copy of a large input dataset in an efficient manner.)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不能在广播之后被修改。
顾名思义,累加器是一种只能通过关联操作进行“加”操作的变量,因此它能够高效的应用于并行操作中。它们能够用来实现counters
和sums
。Spark原生支持数值类型的累加器,开发者可以自己添加支持的类型。如果创建了一个具名的累加器,它可以在spark的UI中显示。这对于理解运行阶段(running stages)的过程有很重要的作用。(注意:这在python中还不被支持)
一个累加器可以通过调用SparkContext.accumulator(v)
方法从一个初始变量v中创建。运行在集群上的任务可以通过add
方法或者使用+=
操作来给它加值。然而,它们无法读取这个值。只有驱动程序可以使用value
方法来读取累加器的值。如下的代码,展示了如何利用累加器将一个数组里面的所有元素相加:
scala> val accum = sc.accumulator(0, "My Accumulator")
accum: spark.Accumulator[Int] = 0
scala> sc.parallelize(Array(1, 2, 3, 4)).foreach(x => accum += x)
...
10/09/29 18:41:08 INFO SparkContext: Tasks finished in 0.317106 s
scala> accum.value
res2: Int = 10
这个例子利用了内置的整数类型累加器。开发者可以利用子类AccumulatorParam创建自己的累加器类型。AccumulatorParam接口有两个方法:zero
方法为你的数据类型提供一个“0 值”(zero value);addInPlace
方法计算两个值的和。例如,假设我们有一个Vector
类代表数学上的向量,我们能够如下定义累加器:
object VectorAccumulatorParam extends AccumulatorParam[Vector] {
def zero(initialValue: Vector): Vector = {
Vector.zeros(initialValue.size)
}
def addInPlace(v1: Vector, v2: Vector): Vector = {
v1 += v2
}
}
// Then, create an Accumulator of this type:
val vecAccum = sc.accumulator(new Vector(...))(VectorAccumulatorParam)
在scala中,Spark支持用更一般的Accumulable接口来累积数据-结果类型和用于累加的元素类型不一样(例如通过收集的元素建立一个列表)。Spark也支持用SparkContext.accumulableCollection
方法累加一般的scala集合类型。
离散流(DStreams)离散流或者DStreams是Spark Streaming提供的基本的抽象,它代表一个连续的数据流。它要么是从源中获取的输入...
在上一章中,我们已经讨论了如何使用Neo4j Native Java API开发和测试Java应用程序。 现在我们将在本章中讨论Neo4j Cypher Java ...
在Mongodb里面存在另一种集群,就是分片技术,可以满足MongoDB数据量大量增长的需求。当MongoDB存储海量的数据时,一台机器可能不...
在MS Access和其他DBMS系统中,查询可以做的不仅仅是显示数据,但它们实际上可以对数据库中的数据执行各种操作。操作查询是可以...