Spark Streaming简介

Spark Streaming是Spark核心API的扩展，可实现实时数据流的可扩展，高吞吐量，容错流式处理。数据可以从诸如Kafka，Flume，Kinesis或TCP套接字的许多来源中获取，并且可以使用由高级函数（如map，reduce，join和window）表达的复杂算法进行处理。另外，处理后的数据可以推送到文件系统、数据库，以及实时仪表板上用于展示。事实上，您还可以在数据流上应用Spark的机器学习和图形处理算法。下图是这些概念的图示：

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

Spark Streaming提供称为离散流或DStream的高级抽象，它表示连续的数据流。可以从诸如Kafka，Flume和Kinesis等来源的输入数据流中创建DStream，或者通过对其他DStream应用高级操作来创建。在内部，DStream表示为一系列RDD。

本指南介绍如何开始使用DStreams编写Spark Streaming程序。您可以在Scala，Java或Python中编写Spark Streaming程序（在Spark 1.2中引入），所有这些都在本指南中有示例介绍，不同语言有对应的代码段。本文仅列出Scala版，其他版本参见：Java&Python Spark Streaming示例

注意：有一些API在Python中是不同的或不可用的。

Spark Streaming示例， Scala版

首先，我们将Spark Streaming类的名称和StreamingContext的一些隐式转换导入到我们的环境中，以便为我们需要的其他类（如DStream）添加有用的方法。 StreamingContext是所有流功能的主要入口点。我们创建一个带有两个执行线程的本地StreamingContext，并且间隔为1秒。

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))

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

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

这行DStream表示将从数据服务器接收的数据流。此DStream中的每个记录都是一行文本。接下来，我们要将空格字符的行分割成单词。

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

flatMap是一个一对多的DStream操作，通过从源DStream中的每个记录生成多个新记录来创建新的DStream。在这种情况下，每一行将被分割成多个单词，并将单词流表示为单词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()

wordsDStream进一步映射（一对一变换）到（word，1）对的DStream，然后reduce获得每批数据中的单词的频率。最后，wordCounts.print（）将打印每秒产生的几个计数。

请注意，当执行这些行时，Spark Streaming仅设置启动时执行的计算，并且尚未开始实际处理。要在完成所有转换设置之后开始处理，我们要调用

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

完整的代码可以在Spark Streaming示例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)
...