五种基于 MapReduce 的并行计算框架介绍及性能测试（4）Spark WordCount 实验
|
- UID
- 1066743
|
五种基于 MapReduce 的并行计算框架介绍及性能测试（4）Spark WordCount 实验
Spark WordCount 实验Spark 与 Hadoop MapReduce 的最大区别是它把所有数据保存在内存中，Hadoop MapReduce 需要从外部存储介质里把数据读入到内存，Spark 不需要这一步骤。它的实现原理与 Hadoop MapReduce 没有太大区别，这里不再重复原理，完整的运行代码如清单 2 所示。
清单 2 . Spark WordCount 示例代码1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
| SparkConf sparkConf = new SparkConf().setAppName("JavaWordCount");
JavaSparkContext ctx = new JavaSparkContext(sparkConf);
JavaRDD<String> lines = ctx.textFile(args[0], Integer.parseInt(args[1]));
JavaRDD<String> words = lines.flatMap(new FlatMapFunction<String, String>() {
@Override
public Iterable<String> call(String s) {
return Arrays.asList(SPACE.split(s));
}
});
//定义 RDD ones
JavaPairRDD<String, Integer> ones = words.mapToPair(new PairFunction<String, String, Integer>() {
@Override
public Tuple2<String, Integer> call(String s) {
return new Tuple2<String, Integer>(s, 1);
}
});
//ones.reduceByKey(func, numPartitions)
JavaPairRDD<String, Integer> counts = ones.reduceByKey(new Function2<Integer, Integer, Integer>() {
@Override
public Integer call(Integer i1, Integer i2) {
return i1 + i2;
}
},10);
//输出 List
List<Tuple2<String, Integer>> output = counts.collect();
Collections.sort(output, new Comparator<Tuple2<String, Integer>>() {
@Override
public int compare(Tuple2<String, Integer> t1,
Tuple2<String, Integer> t2) {
if(t1._2 > t2._2) {
return -1;
} else if(t1._2 < t2._2) {
return 1;
}
return 0;
}
});
|
|
|
|
|
|
|
