Spark 部署及示例代码讲解（4）

look_w

WordCount 的                Java 和 Scala 实现

1	WordCount 的 Java 代码如清单 24 所示。

清单 24.                    WordCount 的 Java 实现代码

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 38 39 40 41 42 43 44 45 46 47 48

public final class JavaWordCount { private static final Pattern SPACE = Pattern.compile(" "); public static void main(String[] args) throws Exception { if (args.length < 1) { System.err.println("Usage: JavaWordCount <file>"); System.exit(1); } //对于所有的 Spark 程序而言，要进行任何操作，首先要创建一个 Spark 的上下文， //在创建上下文的过程中，程序会向集群申请资源以及构建相应的运行环境。 SparkConf sparkConf = new SparkConf().setAppName("JavaWordCount"); JavaSparkContext ctx = new JavaSparkContext(sparkConf); //利用 textFile 接口从文件系统中读入指定的文件，返回一个 RDD 实例对象。 //RDD 的初始创建都是由 SparkContext 来负责的，将内存中的集合或者外部文件系统作为输入源 JavaRDD<String> lines = ctx.textFile(args[0], 1); JavaRDD<String> words = lines.flatMap(                                 new FlatMapFunction<String, String>() { @Override public Iterable<String> call(String s) { return Arrays.asList(SPACE.split(s)); } }); 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); } }); JavaPairRDD<String, Integer> counts = ones.reduceByKey(                new Function2<Integer, Integer, Integer>() { @Override public Integer call(Integer i1, Integer i2) { return i1 + i2; } }); List<Tuple2<String, Integer>> output = counts.collect(); for (Tuple2<?,?> tuple : output) { System.out.println(tuple._1() + ": " + tuple._2()); } ctx.stop(); } }

这里有必要介绍一下这里用到的几个函数。首先是 map 函数，它根据现有的数据集返回一个新的分布式数据集，由每个原元素经过 func 函数转换后组成，这个过程一般叫做转换（transformation）；flatMap 函数类似于 map 函数，但是每一个输入元素，会被映射为 0 到多个输出元素，因此，func 函数的返回值是一个 Seq，而不是单一元素，可以从上面的代码中看出；reduceByKey 函数在一个（K，V) 对的数据集上使用，返回一个（K，V）对的数据集，key 相同的值，都被使用指定的 reduce 函数聚合到一起。                对应的 Scala 版本代码如清单 25 所示。
清单 25.                    WordCount 的 Scala 实现代码

1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19

import org.apache.spark.SparkConf import org.apache.spark.SparkContext import org.apache.spark.SparkContext._ //统计字符出现次数 object WordCount { def main(args: Array[String]) { if (args.length < 1) { System.err.println("Usage: <file>") System.exit(1) } val conf = new SparkConf() val sc = new SparkContext(conf) val line = sc.textFile(args(0))    line.flatMap(_.split(" ")).map((_, )).reduceByKey(_+_).          collect().foreach(println) sc.stop() } }

从清单 24 和 25 对比可以看出，Scala 语言及其简单、轻巧，相对于 Java 语言而言，非常适合于并行计算框架的编写，这也是为什么 Spark 框架是用函数式语言 Scala 写的，而不是 Java 这样的面向对象语言。
运行模式总体上来说，都基于一个相似的工作流程。根本上都是将 Spark 的应用分为任务调度和任务执行两个部分。无论本地模式 or                分布式模式，其内部程序逻辑结构都是类似的，只是其中部分模块有所简化，例如本地模式中，集群管理模块被简化为进程内部的线程池。
所有的 Spark 应用程序都离不开 SparkContext 和 Executor 两部分，Executor 负责执行任务，运行 Executor 的机器称为                Worker 节点，SparkContext 由用户程序启动，通过资源调度模块和 Executor 通信。SparkContext 和 Executor                这两部分的核心代码实现在各种运行模式中都是公用的，在它们之上，根据运行部署模式的不同，包装了不同调度模块以及相关的适配代码。具体来说，以 SparkContext                为程序运行的总入口，在 SparkContext 的初始化过程中，Spark 会分别创建 DAGScheduler 作业调度和 TaskScheduler                任务调度两极调度模块。其中，作业调度模块是基于任务阶段的高层调度模块，它为每个 Spark 作业计算具有依赖关系的多个调度阶段 (通常根据 Shuffle                来划分)，然后为每个阶段构建出一组具体的任务 (通常会考虑数据的本地性等)，然后以 TaskSets(任务组)                的形式提交给任务调度模块来具体执行。而任务调度模块则负责具体启动任务、监控和汇报任务运行情况。
本文是部署及示例代码解释的上篇，在系统文章的中篇会对 Scala 语言进行解释，用 Java 和 Scala 实现相同功能的方式让读者快速掌握 Scala 语言。
结束语 通过本文的学习，读者了解了如何下载、部署 Spark、运行示例代码。此外，深入介绍了运行代码的过程、脚本内容，通过这些介绍力求让读者可以快速地上手                Spark。目前市面上发布的 Spark 中文书籍对于初学者来说大多较为难读懂，作者力求推出一系列 Spark 文章，让读者能够从实际入手的角度来了解                Spark。后续除了应用之外的文章，还会致力于基于 Spark 的系统架构、源代码解释等方面的文章发布。