概述

Spark Streaming是核心Spark API的扩展，对实时数据流地处理具有可扩展，高吞吐量和容错特性。数据可从很多源获取，如Kafka，Flume，Kinesis或者TCP sockets，并且可以使用复杂算法进行处理，用高层函数表示如map，reduce，join和window 。最后，处理结果可以输出到文件系统，数据库或者实时仪表盘。实际上，你可以在数据流上应用machine learning和graph processing算法。

image.png

在Spark内部，工作流程如下。Spark Streaming接收实时输入数据流并且将数据划分为不同的批次，然后交给Spark engine进行处理，按照批次生成最终的结果流。

image.png

Spark Streaming提供了高层抽象，叫做离散流（ discretized stream）或者DStream，代表连续数据流。DStream可以通过Kafka，Flume和Kinesis的输入数据流创建，或者通过在其它DStream上应用高层操作创建。在Spark内部，DStream是一系列RDD。

快速示例

在详细介绍如何写Spark Streaming程序之前，先快速看一下简单的Spark Streaming程序示例。假设我们想计算从数据服务器通过TCP socket发过来的文本数据的单词数量，可参见如下代码。

首先，import Spark Streaming的类和一些隐式转换。StreamingContext是所有流功能的主入口。我们创建一个本地StreamingContext，2个执行线程，批时间间隔为1s。

import org.apache.spark._
import org.apache.spark.streaming._
import org.apache.spark.streaming.StreamingContext._ // not necessary since Spark 1.3

// Create a local StreamingContext with two working thread and batch interval of 1 second.
// The master requires 2 cores to prevent from a starvation scenario.

val conf = new SparkConf().setMaster("local[2]").setAppName("NetworkWordCount")
val ssc = new StreamingContext(conf, Seconds(1))

使用这个context，可以创建DStream，代表TCP源的数据流，指定 hostname (e.g. localhost)和port (e.g. 9999)。

// Create a DStream that will connect to hostname:port, like localhost:9999
val lines = ssc.socketTextStream("localhost", 9999)

lines DStream代表从数据服务器接收的数据流。DStream中的每条记录是文本的一行。接下来，用空格将每行切分成单词。

// Split each line into words
val words = lines.flatMap(_.split(" "))

flatMap是一个一对多的DStream操作，通过从源DStream中的每条记录创建多条新纪录来创建新DStream。我们这个例子中，每行会被切分成多个单词，words DStream代表单词流。接下来，对单词进行计数。

import org.apache.spark.streaming.StreamingContext._ // not necessary since Spark 1.3
// Count each word in each batch
val pairs = words.map(word => (word, 1))
val wordCounts = pairs.reduceByKey(_ + _)

// Print the first ten elements of each RDD generated in this DStream to the console
wordCounts.print()

words DStream进行map（一对一转换）到一个(word, 1) DStream，然后reduce获取每个批次数据的词频。最后，wordCounts.print()会打印其中一些词频。

注意，上面代码行执行时，Spark Streaming只会设置启动时要执行的计算，不会开始真正的处理。要在所有转换完成后开始进行处理，调用以下方法：

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

完整代码可参见NetworkWordCount。

如果你已经下载并且构建了Spark，可以运行下面的示例。需要先运行Netcat（一个小工具，大多数类Unix系统都有）作为数据服务器，如下：

$ nc -lk 9999

然后，在另外一个终端中，启动下面的示例：

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

然后，在运行netcat服务的终端中输入的任意行都会每秒进行计数并打印出来。如下：

# TERMINAL 1:
# Running Netcat

$ nc -lk 9999

hello world



...

# TERMINAL 2: RUNNING NetworkWordCount

$ ./bin/run-example streaming.NetworkWordCount localhost 9999
...
-------------------------------------------
Time: 1357008430000 ms
-------------------------------------------
(hello,1)
(world,1)
...