Apache Spark RDD和Java流

几个月前，我很幸运地参加了一些使用Apache Spark的PoC（概念验证）。在这里，我有机会使用弹性分布式数据集（简称RDD ），转换和操作。

几天后，我意识到虽然Apache Spark和JDK是非常不同的平台，但RDD转换和操作以及流中间操作和终端操作之间存在相似之处。我认为这些相似之处可以帮助初学者（如我* grin * ）开始使用Apache Spark。

Java流	Apache Spark RDD
中级作业	转型
终端操作	行动

请注意，Apache Spark和JDK是
非常不同的平台。 Apache Spark是一个开放源代码集群计算框架，可帮助进行大数据处理和分析。 JDK（Java开发工具包）包括用于开发，调试和监视Java应用程序（而不仅仅是数据处理）的工具。

Java流

让我们从流开始。 Java 8于2014年某个时候发布。可以说，它带来的最重要的功能是Streams API（或简称为Streams）。

创建Stream ，它将提供许多操作，这些操作可以分为两类：

中间，
和终端。

中间操作返回上一个流。这些中间操作可以连接在一起以形成管道。另一方面， 终端操作关闭流管道，并返回结果。

这是一个例子。

Stream.of(1, 2, 3).peek(n -> System.out.println("Peeked at: " + n)).map(n -> n*n).forEach(System.out::println);

运行上面的示例时，它将生成以下输出：

Peeked at: 1
1
Peeked at: 2
4
Peeked at: 3
9

中间操作是懒惰的。直到遇到终端操作，才开始实际执行。在这种情况下，终端操作为forEach() 。因此，我们看不到以下内容。

Peeked at: 1
Peeked at: 2
Peeked at: 3
1
4
9

取而代之的是，我们看到的是： peek() ， map()和forEach()已结合在一起以形成管道。在每遍中，static of()操作从指定的值返回一个元素。然后调用管道： peek()打印字符串“ Peeked at：1”，后跟map() ，并终止于显示数字“ 1”的forEach() 。然后以of()开始的另一遍返回指定值中的下一个元素，然后是peek()和map() ，依此类推。

执行诸如peek()类的中间操作实际上并不会执行任何窥视，而是创建一个新的流，该新流在遍历时将包含初始流的相同元素，但还会执行所提供的操作。

Apache Spark RDD

现在，让我们转到Spark的RDD（弹性分布式数据集）。 Spark处理数据的核心抽象是弹性分布式数据集（RDD）。

RDD只是元素的分布式集合。在Spark中，所有工作都表示为创建新的RDD或调用RDD上的操作以计算结果。在后台，Spark会自动在整个群集中分布RDD中包含的数据，并并行化您对其执行的操作。

创建后，RDD将提供两种类型的操作：

转变，
和行动。

转换从上一个构造新的RDD。另一方面，动作基于RDD计算结果，然后将其返回到驱动程序或将其保存到外部存储系统（例如HDFS）。

这是一个使用Java Streams的大致示例。

SparkConf conf = new SparkConf().setAppName(...);
JavaSparkContext sc = new JavaSparkContext(conf);List<Integer> squares = sc.parallelize(Arrays.asList(1, 2, 3)).map(n -> n*n).collect();System.out.println(squares.toString());// Rough equivalent using Java Streams
List<Integer> squares2 = Stream.of(1, 2, 3).map(n -> n*n).collect(Collectors.toList());System.out.println(squares2.toString());

设置Spark上下文之后，我们调用parallelize() ，该方法从给定的元素列表中创建一个RDD。 map()是一个转换，而collect()是一个动作。像Java中的中间流操作一样，转换会被延迟评估。在此示例中，Spark在看到动作之前不会开始执行对map()的调用中提供的功能。这种方法乍一看可能并不常见，但是在处理大量数据（换句话说就是大数据）时，这很有意义。它允许Spark拆分工作并并行进行。

字数示例

让我们以字数统计为例。在这里，我们有两种实现：一种使用Apache Spark，另一种使用Java Streams。

这是Java Stream版本。

public class WordCountJava {private static final String REGEX = "\\s+";public Map<String, Long> count(URI uri) throws IOException {return Files.lines(Paths.get(uri)).map(line -> line.split(REGEX)).flatMap(Arrays::stream).map(word -> word.toLowerCase()).collect(groupingBy(identity(), TreeMap::new, counting()));}}

在这里，我们逐行读取源文件，并按单词顺序转换每一行（通过map()中间操作）。由于每行都有一个单词序列，并且有很多行，因此可以使用flatMap()将它们转换为单个单词序列。最后，我们根据它们的identity()它们分组（即，字符串的身份就是字符串本身），然后对它们进行计数。

针对包含两行的文本文件进行测试时：

The quick brown fox jumps over the lazy dog
The quick brown fox jumps over the lazy dog

它输出以下地图：

{brown=2, dog=2, fox=2, jumps=2, lazy=2, over=2, quick=2, the=4}

现在，这是Spark版本。

public class WordCountSpark {private static final String REGEX = "\\s+";public List<Tuple2<String, Long>> count(URI uri, JavaSparkContext sc) throws IOException {JavaRDD<String> input = sc.textFile(Paths.get(uri).toString());return input.flatMap(line -> Arrays.asList(line.split(REGEX)).iterator()).map(word -> word.toLowerCase()).mapToPair(word -> new Tuple2<String, Long>(word, 1L)).reduceByKey((x, y) -> (Long) x + (Long) y).sortByKey().collect();}}

当对同一个两行文本文件运行时，它输出以下内容：

[(brown,2), (dog,2), (fox,2), (jumps,2), (lazy,2), (over,2), (quick,2), (the,4)]

为简洁起见，已排除了JavaSparkContext的初始配置。我们从文本文件创建JavaRDD 。值得一提的是，此初始RDD将在文本文件中逐行操作。这就是为什么我们将每一行拆分为单词序列，然后将其flatMap()拆分的原因。然后，我们将一个单词转换为一个计数为一（1）的键值元组，以进行增量计数。完成此操作后，我们将单词（ reduceByKey() ）与上一个RDD中的键值元组进行分组，最后我们以自然顺序对其进行排序。