Spark Streaming编程模型

本节将介绍 Spark Streaming 的编程模型,包括 DStream 的操作流程和使用方法。

DStream 的操作流程

DStream 作为 Spark Streaming 的基础抽象,它代表持续性的数据流。这些数据流既可以通过外部输入源来获取,也可以通过现有的 DStream 的 Transformation 操作来获得。

在内部实现上,DStream 由一组时间序列上连续的 RDD 来表示。如图 1 所示,每个 RDD 都包含了自己特定时间间隔内的数据流。
DStream中在时间轴下生成离散的RDD序列
图 1  DStream中在时间轴下生成离散的RDD序列

如图 2 所示,对 DStream 中数据的各种操作也是映射到内部的 RDD 上来进行的,可以通过 RDD 的 Transformation 生成新的 DStream。这里的执行引擎是 Spark。
DStream中的数据操作流程
图 2  DStream中的数据操作流程

Spark Streaming 使用

作为构建于 Spark 之上的应用框架,Spark Streaming 承袭了 Spark 的编程风格。本节以 Spark Streaming 官方提供的 WordCount 代码为例来介绍 Spark Streaming 的使用方式。
import org.apache.spark._
import org.apache.spark.streaming._
import org.apache.spark.streaming.StreamingContext._

//创建一个拥有两个工作线程,时间片长度为 1 秒的 StreamContext
//主结点需要 2 核以免饥饿状态发生

val conf = new SparkConf().setMaster("local[2]").setAppName("NetworkWordCount");
val ssc = new StreamingContext(conf,Seconds(1))
// 创建连接至 hostname:port 的 DStreamCreate,如 localhost:9999
val lines = ssc.socketTextStream("localhost",9999)
//把每一行分解为单词
val words = lines.flatMap (_.split(""))

import org.apache.spark.streaming.StreamingContext._

//计数每一个时间片内的单词量
val pairs = words.map(word => (word, 1))
val wordCounts =  pairs.reduceByKey(_+_)

//打印该 DStream 生成的每个 RDD 中的前 10 个单词
wordCounts.print()
ssc.start() // 启动计算
ssc.awaitTermination()   //等待计算完成

1. 创建 StreamingContext 对象

Spark Streaming 初始化的主要工作是创建 Streaming Context 对象,通过创建函数的参数指明 Master Server,设定应用名称,指定 Spark Streaming 处理数据的时间间隔等。上述代码可设定应用的名称为 NetworkWordCount,处理数据的时间间隔为 1 秒。

2. 创建 InputDStream

Spark Streaming 需要指明数据源。该实例指明使用 socketTextStream,也就是以 socket 连接作为数据源读取数据。Spark Streaming 支持多种不同的数据源,包括 Kafka、Flume、HDFS/S3、Kinesis、Twitter。

3. 操作 DStream

对于从数据源得到的 DStream,用户可以对其进行各种操作,该实例所示的操作就是一个典型的 WordCount 执行流程。对于当前时间窗口内从数据源得到的数据,首先进行分割,然后利用 map 和 reduceByKey 方法进行计算,最后使用 print() 方法输出结果。

4. 启动 Spark Streaming

之前的所有步骤只是创建了执行流程,程序没有真正连接上数据源,也没有对数据进行任何操作,只是设定好了所有的执行计划,当 ssc.start() 启动后程序才真正进行所有预期的操作。

DStream 的输入源

Spark Streaming 的所有操作都是基于流的,而输入源是这一系列操作的起点。输入 DStream 和 DStream 接收的流都代表输入数据流的来源,Spark Streaming 提供了两种内置数据流来源:基础来源和高级来源。

1. 基础来源

基础来源是在 StreamingContext API 中直接可用的来源,如文件系统、Socket (套接字)等。

前面的例子已经使用了 ssc.socketTextStream() 方法,即通过 TCP 套接字连接,从文本数据中创建一个 DStream。除了套接字之外,StreamingContext 的 API 还提供了从文件和 Akka actors 中创建 DStreams 作为输入源的方法。

Spark Streaming 提供了streamingContext.fileStream(dataDirectory) 方法,该方法可以从任何文件系统(如 HDFS、S3、NFS 等)的文件中读取数据,然后创建一个 DStream。

Spark Streaming 监控 dataDirectory 目录和在该目录下的所有文件的创建处理过程。需要注意的是,文件必须是具有相同的数据格式的,创建的文件必须在 dataDirectory 目录下。对于简单的文本文件,可以使用一个简单的方法 StreamingContext.textFileStream(dataDirectory) 来读取数据。

Spark Streaming 也可以基于自定义 Actors 的流创建 DStream。通过 Akka actors 接收数据流的使用方法是 streamingContext.actorStream(actorProps,actor—name)。

Spark Streaming 使用 streamingContext.queueStream(queueOfRDDs)方法可以创建基于 RDD 队列的 DStream,每个 RDD 队列将被视为 DStream 中的一块数据流进行加工处理。

2.高级来源

局级来源,如 Kafka、Flume、Kinesis、Twitter 等,可以通过额外的实用工具类来创建。高级来源需要外部 non-Spark 库的接口,其中一些有复杂的依赖关系(如 Kafka、Flume)。因此通过这些来源创建 DStreams 需要明确其依赖。例如,如果想创建一个使用 Twitter tweets 的数据的 DStream 流,必须按以下步骤来做。

1)在 sbt 或 maven 工程里添加 spark-streaming-twitter_2.10 依赖。

2)开发:导入 TwitterUtils 包,通过 TwitterUtils.createStream 方法创建一个 DStream。

3)部署:添加所有依赖的 Jar 包,然后部署应用程序。

需要注意的是,这些高级来源一般在 Spark Shell 中不可用,因此基于这些高级来源的应用不能在 Spark Shell 中进行测试。如果必须在 Spark Shell 中使用它们,则需要下载相应的 maven 工程的 Jar 依赖并添加到类路径中。另外,输入 DStream 也可以创建自定义的数据源,需要做的就是实现一个用户定义的接收器。