一、绪论

    上一章节对Spark组件进行了简单的描述，并详细介绍了RDD的内容。本章节主要介绍spark组件之一Spark Streaming的内容。

     Spark Streaming是用来对实时数据进行流式计算的组件。本章节主要从Spark Streaming的原理及其核心DStream两个大方面来介绍Spark Streaming的运行机制。其中实例代码使用的编程语言是Scala语言。

二、原理与架构

  1、流式计算原理 

        从根本上来讲，Spark Streaming并不是完全意义上实现了流式计算。它使用一种“微批次”的架构来实现对数据的流式计算，即把流式计算当成一系列连续的小规模的批处理来对来。也就是“频繁地进行批处理”。

2、流式计算架构

    架构图如图1-1所示：

图1-1 Spark Streaming的高层次架构

    首先，Spark Streaming会从各中输入源中读取数据到接受器上，并将这些数据分组成小的批次。新的批次按均匀的时间间隔创建出来。在每个时间区间开始的时候，一个新的批次就被创建出来，在该区间内收到的数据都会被添加到这个批次中。在时间区间结束时，批次停止增长。每个输入批次都形成了一个RDD，以Spark作业的方式来处理并生成其他的RDD（具体的内容见上一章节的RDD介绍）。处理的结果可以以批处理的方式传给外部系统。

三、DStream

    Spark对数据的抽象表示是RDD，相应地Spark Streaming对流数据的抽象表示是离散化流（discretized stream），叫做DStream。DStream是随时间推移而受到的数据的序列。在其内部，每个时间去收到的数据都作为RDD存在，也就是说DStream是有这些RDD所组成的序列。具体的结构如图2-2所示。

图1-2 DStream结构

1、DStream的创建

    DStream可以从各种输入源来创建，例如Flume、Kafka或者HDFS。这里主要介绍两种比较常用方式：套接字和文件流。

    a、套接字

    从套接字中获取输入流需要从创建StreamingContext开始，因为StreamingContext是流计算功能的主要入口。之后我们需要通过调用socketTextStream()函数来创建出监听某个端口号上收到的数据的DStream。以本地7777端口为例，实现代码如例 1-1所示：

例 1-1 使用套接字作为输入流创建DStream

val ssc=new StreamingContext(conf,Sconds(1))

val lines=ssc.socketTextStream("localhost",7777)

val errorLines=lines.filter(_.contains("error"))

errorLines.print()

    当然，这里只是定义好了要进行的计算，当系统收到数据时就会开始计算。要开始接受数据，就必须显示调用StreamingContext的start()方法。之后，Spark Streaming就会开始吧Spark作业不断交给下面的SparkContext去调度执行。这些执行会在另外一个进程中执行，所以需要调用awaitTermination来等待流计算完成，来防止应用退出。具体代码如例1-2所示。

例 1-2 流式计算的启动

ssc.start()

ssc.awaitTermination()

    b、文件流

    Spark Streaming支持从任意hadoop兼容的文件系统目录中的文件创建数据流。

例 1-3 读取目录中的文本文件流

val logData = ssc.textFileStream(logDirectory)

2、DStream的转化操作

    DStream的转化操作分为无状态和有状态两种。

    a、无状态操作

    在无状态转化操作中，每个批次的处理并不依赖于之前批次的的数据。这种类型的操作会把相对于的简单的RDD转化操作应用到每个批次即DStream中的每一个RDD上。常见的转化操作有map()、filter()、reduceByKey()等。

例 1-4 对DStream使用map()和reduceByKey()操作

val accessLogDStream = logData.map(line => ApacheAccessLog.parseFromLogLine(line))

val ipDStream = accessLogDStream.map(entry => (entry.getIpAddress() , 1))

val ipCountsDStream = ipDStream.reduceByKey((x,y) => x + y)

    需要注意的是reduceByKey()操作会归约每个时间区间内的数据，但是不会归约不同区间之间的数据。

    b、有状态操作

    有状态操作需要使用之前批次的数据或者中间结果来计算当前批次的数据。这是一种跨时间区间的跟踪数据的操作。主要的两种类型是滑动窗口和updateStateByKey()，前者是以一个时间段作为滑动窗口进行操作，后者是用来跟踪每个键的状态变化。

    需要注意的是有状态转化操作需要在StreamingContext中打开检查点机制来保证容错性。

3、DStream的输出操作

    输出操作是指对于流式计算得到的最终结果所要执行的操作，比较常用的有输出到屏幕上和推入外部数据库中。一般用print()来调试输出结果，它会在每个批次中抓取DStream的前十个元素打印出来。Spark Streaming对于DStream的保存是接受一个目录作为参数来存储文件。还支持可选参数来设置文件的后缀名。例1-5就是一个将DStream保存为文本文件的例子，其中"text"参数是可选参数。

例 1-5 将DStream保存为文本文件

ipAddressRequestCount.saveAsTextFiles("outputDir" , "text")

四、参考文献

    [1] Holden Karau ,  Andy Konwinski , Patrick Wendell , Matei Zaharia .Spark快速大数据分析[M].北京:人民邮电出版社,2015.9;