广播变量在最后补充。
今天的示例主要介绍一下累加器的使用。Spark官方文档的Action介绍中有提到 foreach
foreach(func)：Run a function func on each element of the dataset. This is usually done for side effects such as updating an Accumulator or interacting with external storage systems.
Note: modifying variables other than Accumulators outside of theforeach()may result in undefined behavior. See Understanding closures for more details.
意思是说，调用foreach函数可以遍历rdd中的每一个元素，但是在foreach内部，不可以更新除了累加器以外的变量，否则会出现与期望值不符的结果。
例如，你想做一个简单求和计数，首先声明一个变量，然后在foreach内部增值。请注意，这样是行不通的。
官方文档中这样介绍：By default, when Spark runs a function in parallel as a set of tasks on different nodes, it ships a copy of each variable used in the function to each task. 默认情况下，当Spark以不同节点上的一组任务并行运行一个函数时，它会将该函数中使用的每个变量的副本发送给每个任务。
为了解决这个问题，除了之前示例中用过的reduce函数外，还可以引入累加器。
package com.yzy.spark;
import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.function.VoidFunction;
import org.apache.spark.util.LongAccumulator;
import java.util.Arrays;
import java.util.List;
public class demo3 {
private static String appName = "spark.demo";
private static String master = "local[*]";
public static void main(String[] args) {
JavaSparkContext sc = null;
try {
//初始化 JavaSparkContext
SparkConf conf = new SparkConf().setAppName(appName).setMaster(master);
sc = new JavaSparkContext(conf);
//构造数据源
List<Integer> data = Arrays.asList(1, 2, 3, 4, 5);
//并行化创建rdd
JavaRDD<Integer> rdd = sc.parallelize(data);
//累加器
final LongAccumulator accumulator = sc.sc().longAccumulator();
rdd.foreach(new VoidFunction<Integer>() {
public void call(Integer integer) throws Exception {
accumulator.add(integer);
}
});
System.out.println("执行结果：" + accumulator.value());
} catch (Exception e) {
e.printStackTrace();
} finally {
if (sc != null) {
sc.close();
}
}
}
}
执行结果
//省略若干行
18/06/25 14:16:24 INFO TaskSetManager: Finished task 3.0 in stage 0.0 (TID 3) in 89 ms on localhost (executor driver) (1/4)
18/06/25 14:16:24 INFO TaskSetManager: Finished task 2.0 in stage 0.0 (TID 2) in 92 ms on localhost (executor driver) (2/4)
18/06/25 14:16:24 INFO TaskSetManager: Finished task 1.0 in stage 0.0 (TID 1) in 94 ms on localhost (executor driver) (3/4)
18/06/25 14:16:24 INFO TaskSetManager: Finished task 0.0 in stage 0.0 (TID 0) in 113 ms on localhost (executor driver) (4/4)
18/06/25 14:16:24 INFO TaskSchedulerImpl: Removed TaskSet 0.0, whose tasks have all completed, from pool
18/06/25 14:16:24 INFO DAGScheduler: ResultStage 0 (foreach at demo3.java:35) finished in 0.298 s
执行结果：15
累加器的使用比较简单，但是需要注意的是，以下两种情况会造成累加器计数出错
第一种：计数为0
//累加器
final LongAccumulator accumulator = sc.sc().longAccumulator();
//map && reduce
JavaRDD<Integer> javaRDD = rdd.map(new Function<Integer, Integer>() {
public Integer call(Integer integer) throws Exception {
accumulator.add(integer);
return integer;
}
});
System.out.println("执行结果：" + accumulator.value());
-----------------------
执行结果：0
可以看到，我们把累加器放到map中执行的时候，结果输出为0。这个很容易想到，因为map是转换操作，是惰性的。rdd没有启动Action，map不会执行，因此输出为0。
第二种：计数翻倍
//累加器
final LongAccumulator accumulator = sc.sc().longAccumulator();
//map && reduce
JavaRDD<Integer> javaRDD = rdd.map(new Function<Integer, Integer>() {
public Integer call(Integer integer) throws Exception {
accumulator.add(integer);
return integer;
}
});
//reduce
javaRDD.reduce(new Function2<Integer, Integer, Integer>() {
public Integer call(Integer integer, Integer integer2) throws Exception {
return integer + integer2;
}
});
//count
javaRDD.count();
System.out.println("执行结果：" + accumulator.value());
-----------------------
执行结果：30
当执行reduce操作时，启动作业，map开始执行，累加器的值应为15。但是rdd又执行了一次count操作，count也是Action，map操作再次执行，由于累加器此时不为0，所以最终结果翻倍，输出30。
解决这个问题的方法是：开启rdd 缓存
//reduce，rdd缓存
javaRDD.cache().reduce(new Function2<Integer, Integer, Integer>() {
public Integer call(Integer integer, Integer integer2) throws Exception {
return integer + integer2;
}
});
即rdd 第一次执行action算子的时候开启cache缓存，后面的action算子不会触发map执行。
实际上rdd缓存函数是persist函数，它支持不同类型的缓存策略，cache()等同于persist(StorageLevel.MEMORY_ONLY())。其他缓存级别可参考官网。
补充：缓存删除Spark会自动监视每个节点上的缓存使用情况，并以最近最少使用（LRU）的方式删除旧数据分区。如果您想手动删除RDD而不是等待它退出缓存，请使用该RDD.unpersist()方法。
广播变量
官方介绍：Broadcast variables allow the programmer 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 also attempts to distribute broadcast variables using efficient broadcast algorithms to reduce communication cost.广播变量是Spark中另一种共享变量，允许程序将一个只读的变量发送到Executor，一个Executor只需要在第一个Task启动时，获得一份Broadcast数据，之后的Task都从本节点的BlockManager中获取相关数据。
使用方法非常简单
//定义一个String类型的变量
final Broadcast<String> broadcast = sc.broadcast("broadcast");
...
//rdd中使用
broadcast.value()
网友评论