Spark Streaming-1：Spark Streaming编程指南

素颜马尾好姑娘i 2022-07-11 07:38 299阅读 0赞

翻译：http://spark.apache.org/docs/latest/streaming-programming-guide.html

**概述**

Spark Streaming是核心Spark API的扩展，能够对实时数据流进行可扩展，高吞吐量，容错的流处理。数据可以从诸如Kafka，Flume，Kinesis或TCP套接字的许多源中提取，并且可以使用由诸如`map`, `reduce`, `join` and `window`的高级功能表示的复杂算法来处理。最后，处理的数据可以推送到文件系统，数据库和活动仪表板。事实上，你可以对数据流应用Spark的机器学习和图形处理算法。

![Spark Streaming][]

在内部，它的工作原理如下。 Spark Streaming接收实时输入数据流，并将数据分成批次，然后由Spark引擎处理以批量生成最终结果流。

![Spark Streaming][Spark Streaming 1]

Spark Streaming提供了一种称为离散流或DStream的高级抽象，表示连续的数据流。 DStreams可以从来源（例如Kafka，Flume和Kinesis）的输入数据流创建，也可以通过对其他DStreams应用高级操作来创建。在内部，DStream表示为RDD序列。

本指南介绍如何开始使用DStreams编写Spark Streaming程序。您可以在Scala，Java或Python（在Spark 1.2中介绍）中编写Spark Streaming程序，本指南中介绍了这些程序。您将在本指南中找到标签，您可以在不同语言的代码段之间进行选择。

注意：Python中有几个API不同或不可用。在本指南中，您将看到标记Python API突出这些差异。

**示例**

在我们进入如何编写自己的Spark Streaming程序的细节之前，让我们快速看看一个简单的Spark Streaming程序的样子。假设我们要计算从侦听TCP套接字的数据服务器接收的文本数据中的字数。所有你需要做的是如下。

首先，我们创建一个JavaStreamingContext对象，这是所有流功能的主要入口点。我们创建一个具有两个执行线程的本地StreamingContext，批处理间隔为1秒。

import org.apache.spark.*;
    import org.apache.spark.api.java.function.*;
    import org.apache.spark.streaming.*;
    import org.apache.spark.streaming.api.java.*;
    import scala.Tuple2;
    
    // Create a local StreamingContext with two working thread and batch interval of 1 second
    SparkConf conf = new SparkConf().setMaster("local[2]").setAppName("NetworkWordCount");
    JavaStreamingContext jssc = new JavaStreamingContext(conf, Durations.seconds(1));

使用此context，我们可以创建一个DStream，它表示来自TCP源的流数据，指定为主机名（例如localhost）和端口（例如9999）。

JavaReceiverInputDStream<String> lines = jssc.socketTextStream("localhost", 9999);

此行DStream表示将从数据服务器接收的数据流。此流中的每个记录都是一行文本。然后，我们要将行按空格拆分为单词。

// Split each line into words
 JavaDStream<String> words = lines.flatMap(
 new FlatMapFunction<String, String>() { 
 @Override public Iterator<String> call(String x) { 
 return Arrays.asList(x.split(" ")).iterator();
 }
 });

flatMap是一个DStream操作，通过从源DStream中的每个记录生成多个新记录来创建新的DStream。在这种情况下，每一行将被拆分成多个字，并且字流被表示为字节DStream。注意，我们使用FlatMapFunction对象定义了转换。正如我们将会发现，在Java API中有许多这样的方便类有助于定义DStream转换。

接下来，我们要计算这些字。

// Count each word in each batch
 JavaPairDStream<String, Integer> pairs = words.mapToPair(
 new PairFunction<String, String, Integer>() { 
 @Override public Tuple2<String, Integer> call(String s) { 
 return new Tuple2<>(s, 1);
 }
 });
 JavaPairDStream<String, Integer> wordCounts = pairs.reduceByKey(
 new Function2<Integer, Integer, Integer>() { 
 @Override public Integer call(Integer i1, Integer i2) { 
 return i1 + i2;
 }
 });
 
 // Print the first ten elements of each RDD generated in this DStream to the console
 wordCounts.print();

使用PairFunction对象，将DStream字进一步映射（一对一变换）到（字，1）对的DStream。然后，使用Function2对象来减少每一批数据中的字的频率。最后，wordCounts.print（）将打印每秒生成的几个计数。

注意，当执行这些行时，Spark Streaming仅设置它在启动后将执行的计算，并且还没有开始真正的处理。要在所有变换设置完成后开始处理，我们最后调用start方法。

jssc.start();              // Start the computation
    jssc.awaitTermination();   // Wait for the computation to terminate

完整的代码可以在Spark Streaming示例JavaNetworkWordCount中找到。

如果您已经下载并构建了Spark，则可以按如下所示运行此示例。您将首先需要运行Netcat（在大多数类Unix系统中找到的小实用程序）作为数据服务器

$ nc -lk 9999

然后，在另一个终端中，您可以使用启动示例

$ ./bin/run-example streaming.JavaNetworkWordCount localhost 9999

然后，在运行netcat服务器的终端中键入的任何行将被计数并每秒在屏幕上打印。它将看起来像下面。

# TERMINAL 1:
    # Running Netcat
    
    $ nc -lk 9999
    
    hello world

# TERMINAL 2: RUNNING JavaNetworkWordCount
    
    $ ./bin/run-example streaming.JavaNetworkWordCount localhost 9999
    ...
    -------------------------------------------
    Time: 1357008430000 ms
    -------------------------------------------
    (hello,1)
    (world,1)

**基本概念**

与Spark类似，Spark Streaming通过Maven Central提供。要编写自己的Spark Streaming程序，您必须将以下依赖项添加到SBT或Maven项目中。

<dependency>
 <groupId>org.apache.spark</groupId>
 <artifactId>spark-streaming_2.11</artifactId>
 <version>2.1.0</version>
 </dependency>

对于从Spark Streaming核心API中不存在的来源（如Kafka，Flume和Kinesis）获取数据，您必须将相应的artifact  spark-streaming-xyz\_2.11添加到依赖项。例如，一些常见的如下。

*For an up-to-date list, please refer to the [Maven repository][] for the full list of supported sources and artifacts.*

**初始化StreamingContext**

import org.apache.spark.*;
    import org.apache.spark.streaming.api.java.*;
    
    SparkConf conf = new SparkConf().setAppName(appName).setMaster(master);
    JavaStreamingContext ssc = new JavaStreamingContext(conf, new Duration(1000));

appName参数是应用程序在集群UI上显示的名称。 master是Spark，Mesos或YARN集群URL，或者是一个在本地模式下运行的特殊“local \[\*\]”字符串。

在实践中，当在集群上运行时，您不要想在程序中硬编码主程序，而是使用spark-submit启动应用程序并接收它。但是，对于本地测试和单元测试，您可以传递“local \[\*\]”来运行Spark Streaming in-process（检测本地系统中的核心数）。注意，这里内部创建一个SparkContext（所有Spark功能的起始点），可以作为ssc.sparkContext访问。

批处理间隔必须根据应用程序和可用群集资源的延迟要求进行设置。有关更多详细信息，请参阅“性能调优”部分。

StreamingContext对象也可以从现有的SparkContext对象创建。

import org.apache.spark.streaming._
    
    val sc = ...                // existing SparkContext
    val ssc = new StreamingContext(sc, Seconds(1))

**定义context 后，您必须执行以下操作。**

通过创建输入DStreams定义输入源。

通过对DStreams应用变换和输出操作来定义流计算。

开始接收数据并使用streamingContext.start（）处理它。

等待处理停止（手动或由于任何错误）使用streamingContext.awaitTermination（）。

可以使用streamingContext.stop（）手动停止处理。

**注意事项：**

一旦上下文已经开始，则不能设置或添加新的流计算。

上下文停止后，无法重新启动。

只有一个StreamingContext可以在JVM中同时处于活动状态。

stop（）在StreamingContext也停止SparkContext。要仅停止StreamingContext，请将名为stopSparkContext的stop（）的可选参数设置为false。

SparkContext可以重用于创建多个StreamingContext，只要先前的StreamingContext在创建下一个StreamingContext之前停止（不停止SparkContext）即可。

## Discretized Streams (DStreams) ##

离散流或DStream是由Spark Streaming提供的基本抽象。它表示连续的数据流，从源接收的输入数据流或通过变换输入流生成的处理数据流。在内部，DStream由连续的RDD系列表示，这是Spark对不可变，分布式数据集的抽象（更多细节参见Spark编程指南）。 DStream中的每个RDD包含来自某个间隔的数据，如下图所示。

![Spark Streaming][Spark Streaming 2]

**对DStream应用的任何操作都转换为基础RDD上的操作。例如，在将行流转换为字的较早示例中，将flatMap操作应用于行DStream中的每个RDD以生成字DStream的RDD。如下图所示。**

**![Spark Streaming][Spark Streaming 3]**

这些潜在的RDD转换由Spark引擎计算。 DStream操作隐藏了大多数这些细节，并为开发人员提供了更高级别的API以方便使用。这些操作将在后面的章节中详细讨论。

## Input DStreams and Receivers ##

输入DStreams是表示从流源接收的输入数据流的DStream。在快速示例中，lines是一个输入DStream，因为它表示从netcat服务器接收的数据流。每个输入DStream（文件流除外，本节稍后将讨论）与Receiver（Scala doc，Java doc）对象相关联，该对象从源接收数据并将其存储在Spark的内存中进行处理。

Spark Streaming提供两种类型的内置流媒体源。

基本源：StreamingContext API中直接可用的源。示例：文件系统和套接字连接。

高级源：Kafka，Flume，Kinesis等源可通过额外的实用程序类获得。这些需要链接到额外的依赖关系，如链接部分所讨论的。

请注意，

如果您想在流应用程序中并行接收多个数据流，您可以创建多个输入DStream（在“性能调优”部分中进一步讨论）。这将创建多个接收器，这将同时接收多个数据流。但请注意，Spark worker / executor是一个长期运行的任务，因此它占用了分配给Spark Streaming应用程序的核心之一。

因此，重要的是要记住，Spark Streaming应用程序需要分配足够的核心（或线程，如果在本地运行）以处理接收的数据，以及运行接收器。

**基本来源**

File Streams:

对于从与HDFS API兼容的任何文件系统（即HDFS，S3，NFS等）上的文件读取数据，DStream可以创建为：

streamingContext.fileStream<KeyClass, ValueClass, InputFormatClass>(dataDirectory);

Spark Streaming将监视目录dataDirectory并处理在该目录中创建的任何文件（暂不支持的嵌套目录中写入的文件）。

注意文件必须具有相同的数据格式。

这些文件必须在dataDirectory中通过原子方式移动或重命名到数据目录中创建。移动后，不能更改文件。

因此，如果文件被连续附加，新的数据将不会被读取。对于简单的文本文件，有一个更简单的方streamingContext.textFileStream（dataDirectory）。文件流不需要运行接收器，因此不需要分配内核。

Streams based on Custom Receivers:

可以使用通过自定义接收器接收的数据流创建DStreams。有关详细信息，请参阅自定义接收器指南。

Queue of RDDs as a Stream:要使用测试数据测试Spark Streaming应用程序，还可以使用streamingContext.queueStream（queueOfRDDs）基于RDD队列创建DStream。推送到队列中的每个RDD将被视为DStream中的一批数据，并像流一样处理。

**高级源**

这类源需要与外部非Spark库连接，其中一些具有复杂的依赖性（例如，Kafka和Flume）。因此，为了尽量减少与依赖关系的版本冲突相关的问题，从这些源创建DStreams的功能已经移动到可以在必要时显式链接的单独的库。请注意，这些高级源在Spark外壳中不可用，因此基于这些高级源的应用程序无法在shell中测试。如果你真的想在Spark shell中使用它们，那么你必须下载相应的Maven工件的JAR及其依赖项，并将其添加到类路径中。

这些高级来源中的一些如下。

*  Kafka: Spark Streaming 2.1.0 is compatible with Kafka broker versions 0.8.2.1 or higher. See the [Kafka Integration Guide][] for more details.
 *  Flume: Spark Streaming 2.1.0 is compatible with Flume 1.6.0. See the [Flume Integration Guide][] for more details.
 *  Kinesis: Spark Streaming 2.1.0 is compatible with Kinesis Client Library 1.2.1. See the [Kinesis Integration Guide][] for more details.

### 自定义来源    ###

基于它们的可靠性，可以存在两种数据源。

源（如Kafka和Flume）允许传输的数据被确认。如果从这些可靠源接收数据的系统正确地确认所接收的数据，则可以确保没有数据将由于任何种类的故障而丢失。这导致两种接收器：

可靠的接收器 - 当数据已经被接收并存储在带复制的Spark中时，可靠的接收器正确地向可靠的源发送确认。

不可靠的接收器 - 不可靠的接收器不向源发送确认。这可以用于不支持确认的源，或者当不需要或者需要进行确认的复杂性时，用于可靠源。有关如何编写可靠接收器的详细信息，请参阅自定义接收器指南。

**DStreams的转换**

类似于RDD，转换允许来自输入DStream的数据被修改。 DStreams支持在正常的Spark RDD上可用的许多转换。一些常见的如下。

<table style="width:948px; border-collapse:collapse; border-spacing:0px; border:1px solid rgb(238,238,238); color:rgb(85,85,85); font-family:&quot;microsoft yahei&quot;; font-size:14px"> 
 <thead style=""> 
 <tr style=""> 
 <th style="padding:8px; line-height:20px; vertical-align:top; border:1px solid rgb(238,238,238)"> Transformation</th> 
 <th align="center" style="padding:8px; line-height:20px; vertical-align:top; border:1px solid rgb(238,238,238)"> Meaning</th> 
 </tr> 
 </thead> 
 <tbody style=""> 
 <tr style=""> 
 <td style="padding:8px; line-height:20px; vertical-align:top; border:1px solid rgb(238,238,238)"> map(func)</td> 
 <td align="center" style="padding:8px; line-height:20px; vertical-align:top; border:1px solid rgb(238,238,238)"> 对DStream中的各个元素进行func函数操作，然后返回一个新的DStream.</td> 
 </tr> 
 <tr style=""> 
 <td style="padding:8px; line-height:20px; vertical-align:top; border:1px solid rgb(238,238,238)"> flatMap(func)</td> 
 <td align="center" style="padding:8px; line-height:20px; vertical-align:top; border:1px solid rgb(238,238,238)"> 与map方法类似，只不过各个输入项可以被输出为零个或多个输出项</td> 
 </tr> 
 <tr style=""> 
 <td style="padding:8px; line-height:20px; vertical-align:top; border:1px solid rgb(238,238,238)"> filter(func)</td> 
 <td align="center" style="padding:8px; line-height:20px; vertical-align:top; border:1px solid rgb(238,238,238)"> 过滤出所有函数func返回值为true的DStream元素并返回一个新的DStream</td> 
 </tr> 
 <tr style=""> 
 <td style="padding:8px; line-height:20px; vertical-align:top; border:1px solid rgb(238,238,238)"> repartition(numPartitions)</td> 
 <td align="center" style="padding:8px; line-height:20px; vertical-align:top; border:1px solid rgb(238,238,238)"> 增加或减少DStream中的分区数，从而改变DStream的并行度</td> 
 </tr> 
 <tr style=""> 
 <td style="padding:8px; line-height:20px; vertical-align:top; border:1px solid rgb(238,238,238)"> union(otherStream)</td> 
 <td align="center" style="padding:8px; line-height:20px; vertical-align:top; border:1px solid rgb(238,238,238)"> 将源DStream和输入参数为otherDStream的元素合并，并返回一个新的DStream.</td> 
 </tr> 
 <tr style=""> 
 <td style="padding:8px; line-height:20px; vertical-align:top; border:1px solid rgb(238,238,238)"> count()</td> 
 <td align="center" style="padding:8px; line-height:20px; vertical-align:top; border:1px solid rgb(238,238,238)"> 通过对DStreaim中的各个RDD中的元素进行计数，然后返回只有一个元素的RDD构成的DStream</td> 
 </tr> 
 <tr style=""> 
 <td style="padding:8px; line-height:20px; vertical-align:top; border:1px solid rgb(238,238,238)"> reduce(func)</td> 
 <td align="center" style="padding:8px; line-height:20px; vertical-align:top; border:1px solid rgb(238,238,238)"> 对源DStream中的各个RDD中的元素利用func进行聚合操作，然后返回只有一个元素的RDD构成的新的DStream.</td> 
 </tr> 
 <tr style=""> 
 <td style="padding:8px; line-height:20px; vertical-align:top; border:1px solid rgb(238,238,238)"> countByValue()</td> 
 <td align="center" style="padding:8px; line-height:20px; vertical-align:top; border:1px solid rgb(238,238,238)"> 对于元素类型为K的DStream，返回一个元素为（K,Long）键值对形式的新的DStream，Long对应的值为源DStream中各个RDD的key出现的次数</td> 
 </tr> 
 <tr style=""> 
 <td style="padding:8px; line-height:20px; vertical-align:top; border:1px solid rgb(238,238,238)"> reduceByKey(func, [numTasks])</td> 
 <td align="center" style="padding:8px; line-height:20px; vertical-align:top; border:1px solid rgb(238,238,238)"> 利用func函数对源DStream中的key进行聚合操作，然后返回新的（K，V）对构成的DStream</td> 
 </tr> 
 <tr style=""> 
 <td style="padding:8px; line-height:20px; vertical-align:top; border:1px solid rgb(238,238,238)"> join(otherStream, [numTasks])</td> 
 <td align="center" style="padding:8px; line-height:20px; vertical-align:top; border:1px solid rgb(238,238,238)"> 输入为（K,V)、（K,W）类型的DStream，返回一个新的（K，（V，W）类型的DStream</td> 
 </tr> 
 <tr style=""> 
 <td style="padding:8px; line-height:20px; vertical-align:top; border:1px solid rgb(238,238,238)"> cogroup(otherStream, [numTasks])</td> 
 <td align="center" style="padding:8px; line-height:20px; vertical-align:top; border:1px solid rgb(238,238,238)"> 输入为（K,V)、（K,W）类型的DStream，返回一个新的 (K, Seq[V], Seq[W]) 元组类型的DStream</td> 
 </tr> 
 <tr style=""> 
 <td style="padding:8px; line-height:20px; vertical-align:top; border:1px solid rgb(238,238,238)"> transform(func)</td> 
 <td align="center" style="padding:8px; line-height:20px; vertical-align:top; border:1px solid rgb(238,238,238)"> 通过RDD-to-RDD函数作用于源码DStream中的各个RDD，可以是任意的RDD操作，从而返回一个新的RDD</td> 
 </tr> 
 <tr style=""> 
 <td style="padding:8px; line-height:20px; vertical-align:top; border:1px solid rgb(238,238,238)"> updateStateByKey(func)</td> 
 <td align="center" style="padding:8px; line-height:20px; vertical-align:top; border:1px solid rgb(238,238,238)"> 根据于key的前置状态和key的新值，对key进行更新，返回一个新状态的DStream</td> 
 </tr> 
 </tbody> 
</table>

**DStreams的输出操作**

输出操作允许DStream的数据被推送到外部系统，如数据库或文件系统。由于输出操作实际上允许转换的数据被外部系统消耗，它们触发所有DStream转换的实际执行（类似于RDD的动作）。目前，定义了以下输出操作：

<table style="max-width:100%; background-color:transparent; border-collapse:collapse; border-spacing:0px; width:914px; margin-bottom:20px; color:rgb(29,31,34); font-family:&quot;Helvetica Neue&quot;,Helvetica,Arial,sans-serif; font-size:14px"> 
 <tbody> 
 <tr> 
 <th style="padding:8px; line-height:20px; text-align:left; vertical-align:top; border-top:1px solid rgb(221,221,221); width:258px"> Output Operation</th> 
 <th style="padding:8px; line-height:20px; text-align:left; vertical-align:top; border-top:1px solid rgb(221,221,221)"> Meaning</th> 
 </tr> 
 <tr> 
 <td style="padding:8px; line-height:20px; vertical-align:top; border-top:1px solid rgb(221,221,221)"> print()</td> 
 <td style="padding:8px; line-height:20px; vertical-align:top; border-top:1px solid rgb(221,221,221)"> Prints the first ten elements of every batch of data in a DStream on the driver node running the streaming application. This is useful for development and debugging.&nbsp; Python API&nbsp;This is called&nbsp;pprint()&nbsp;in the Python API.</td> 
 </tr> 
 <tr> 
 <td style="padding:8px; line-height:20px; vertical-align:top; border-top:1px solid rgb(221,221,221)"> saveAsTextFiles(prefix, [suffix])</td> 
 <td style="padding:8px; line-height:20px; vertical-align:top; border-top:1px solid rgb(221,221,221)"> Save this DStream's contents as text files. The file name at each batch interval is generated based on&nbsp;prefix&nbsp;and&nbsp;suffix:&nbsp;"prefix-TIME_IN_MS[.suffix]".</td> 
 </tr> 
 <tr> 
 <td style="padding:8px; line-height:20px; vertical-align:top; border-top:1px solid rgb(221,221,221)"> saveAsObjectFiles(prefix, [suffix])</td> 
 <td style="padding:8px; line-height:20px; vertical-align:top; border-top:1px solid rgb(221,221,221)"> Save this DStream's contents as&nbsp;<code style="padding:0px; font-family:Menlo,&quot;Lucida Console&quot;,monospace; color:rgb(68,68,68); border:none">SequenceFiles</code>&nbsp;of serialized Java objects. The file name at each batch interval is generated based on&nbsp;prefix&nbsp;and&nbsp;suffix:&nbsp;"prefix-TIME_IN_MS[.suffix]".&nbsp; Python API&nbsp;This is not available in the Python API.</td> 
 </tr> 
 <tr> 
 <td style="padding:8px; line-height:20px; vertical-align:top; border-top:1px solid rgb(221,221,221)"> saveAsHadoopFiles(prefix, [suffix])</td> 
 <td style="padding:8px; line-height:20px; vertical-align:top; border-top:1px solid rgb(221,221,221)"> Save this DStream's contents as Hadoop files. The file name at each batch interval is generated based on&nbsp;prefix&nbsp;and&nbsp;suffix:&nbsp;"prefix-TIME_IN_MS[.suffix]".&nbsp; Python API&nbsp;This is not available in the Python API.</td> 
 </tr> 
 <tr> 
 <td style="padding:8px; line-height:20px; vertical-align:top; border-top:1px solid rgb(221,221,221)"> foreachRDD(func)</td> 
 <td style="padding:8px; line-height:20px; vertical-align:top; border-top:1px solid rgb(221,221,221)"> The most generic output operator that applies a function,&nbsp;func, to each RDD generated from the stream. This function should push the data in each RDD to an external system, such as saving the RDD to files, or writing it over the network to a database. Note that the function&nbsp;func&nbsp;is executed in the driver process running the streaming application, and will usually have RDD actions in it that will force the computation of the streaming RDDs.</td> 
 </tr> 
 </tbody> 
</table>

[Spark Streaming]: /images/20220711/bf84ebf348554d9d999fb981f257ff8f.png
[Spark Streaming 1]: /images/20220711/9106be0d5845448ab255a59c7c9c599a.png
[Maven repository]: http://search.maven.org/#search%7Cga%7C1%7Cg%3A%22org.apache.spark%22%20AND%20v%3A%222.1.0%22
[Spark Streaming 2]: /images/20220711/3f1536c339b9469790b18b272b0da51e.png
[Spark Streaming 3]: /images/20220711/8d4bb356e7b24f41849c42c2b9633572.png
[Kafka Integration Guide]: http://spark.apache.org/docs/latest/streaming-kafka-integration.html
[Flume Integration Guide]: http://spark.apache.org/docs/latest/streaming-flume-integration.html
[Kinesis Integration Guide]: http://spark.apache.org/docs/latest/streaming-kinesis-integration.html