spark之路第七课——streaming(1)

DStream

DStream在spark中表示一系列的数据流，其中的内容就是一个个的RDD数据集，在DStream上的操作会被转化为操作其内部的RDD。

Input DStream

input DStream是从各种数据源接收到的流，每个input DStream都有一个receiver对象用来从数据源接收数据并且将数据存放在内存中。所以每个input DStream只会接收一个单独的数据流，但可以创建多个input DStream以便能并行的接收到多个数据流。
接收器是作为一个长时间运行的任务运行在spark的worker或executor中，因此一个接收器会占用一个cpu，所以spark streaming应用需要分配足够的cpu来处理接收到的数据。
注意：

基本数据源

上面例子使用了tcp作为数据源，另外还有很多其他的基本数据源：

streamingContext.fileStream\[keyClass, valueClass, inputFormatClass\](dataDirectory)

spark streaming会从dataDirectory文件夹读取所有文件（除该文件夹中子文件夹里的文件外），需要注意这些文件必须有相同的数据格式；这些文件必须以原子移动或重命名的方式在这个文件夹内创建；一旦这些文件被创建，那么它就不能被改变了，所以以后在这些文件里添加新的内容时这些新的内容将不会被读取。

高级数据源

高级数据源包括之前提到的kafka，flume，twitter等，比如要使用twitter的数据流创建DStream，需要做以下事情：
1.将spark-streaming-twitter_2.10添加到sbt或maven中
2.导入TwitterUtils类并使用TwitterUtils.createStream(sparkStreamingContext)方法来创建DStream
3.生成jar包部署改程序

DStream转换

DStream支持的常见转换方式有以下几种：

下面详细讲下最后两个装换方法：

UpdateStateByKey

这个方法允许获取数据新的状态，需要以下两个步骤：
1.首先需要定义数据的初始状态：状态可以为多种数据类型
2.定义状态改变函数：该函数利用数据之前的状态和stream中新的值返回新的状态
下面的例子中状态改变函数将计数作为状态，其中pairs DStream是（word,1）格式：

def updateFunction(newValues: Seq[Int], runningCount: Option[Int]): Option[Int] = {
    val newCount = ...  // add the new values with the previous running count to get the new count
    Some(newCount)
}
val runningCounts = pairs.updateStateByKey[Int](updateFunction _)

上面的例子中每个word都会传递到状态改变函数中处理，传递的newValues就是一系列的1，所以上述例子也可以用来计算每个word的次数。

转换操作

转换操作允许DStream执行RDD-RDD函数，转换操作可以用于任何RDD上。例如当你想实时将输入数据与之前计算好的数据进行合并执行filter操作，可以：

val spamInfoRDD = ssc.sparkContext.newAPIHadoopRDD(...) // RDD containing spam information
val cleanedDStream = wordCounts.transform(rdd => {
  rdd.join(spamInfoRDD).filter(...) // join data stream with spam information to do data cleaning
  ...
})

窗口操作

spark streaming也提供了窗口计算，它允许在滑动窗口上的数据进行转换操作。下图介绍了滑动窗口： 上图显示，每个时刻窗口会在源DStream上滑动，在这个窗口里的RDD会被合并和进行相应的操作生成窗口DStream。在上图的情形下，相关操作会在最后3个时间单元的数据上进行，并且滑动2个时间单元，所以窗口操作需要两个参数：

// Reduce last 30 seconds of data, every 10 seconds
val windowedWordCounts = pairs.reduceByKeyAndWindow((a:Int,b:Int) => (a + b), Seconds(30), Seconds(10))

例子中用到了reduceByKeyAndWindow操作，其他的一些常见窗口操作如下：

DStream输出操作

输出操作允许DStream的数据放入外置系统如数据库或文件系统，由于输出操作需要将最终数据放入外置系统中，所以执行输出操作的时候会开始真正执行之前的转换操作。下表是一些输出操作：

foreachRDD的设计模式

dstream.foreachRDD是非常有用的操作，但使用它需要避免如下的几个错误：

dstream.foreachRDD(rdd => {
      val connection = createNewConnection()  // executed at the driver
      rdd.foreach(record => {
          connection.send(record) // executed at the worker
      })
  })

这种方式是不正确的，因为这需要将这个连接序列化以便将它从driver传送到worker中，这样的连接对象是不能在机器之间进行传送的。上面会发生序列化错误（连接对象不能被序列化）和初始化错误（连接对象需要在worker端进行初始化）等错误。正确的做法是在worker端进行创建连接对象。

dstream.foreachRDD(rdd => {
      rdd.foreach(record => {
          val connection = createNewConnection()
          connection.send(record)
          connection.close()
      })
  })

创建连接对象是需要时间和资源的，所以这种方式会明显的增加不必要的资源消耗和降低机器的吞吐能力，一个更好的解决方案是使用rdd.foreachPartition-创建一个单独的连接对象并且通过这个连接发送一个RDD中所有的数据：

dstream.foreachRDD(rdd => {
      rdd.foreachPartition(partitionOfRecords => {
          val connection = createNewConnection()
          partitionOfRecords.foreach(record => connection.send(record))
          connection.close()
      })
  })

但这种方法在多个RDD分区中还是会创建多个连接。

dstream.foreachRDD(rdd => {
      rdd.foreachPartition(partitionOfRecords => {
          // ConnectionPool is a static, lazily initialized pool of connections
          val connection = ConnectionPool.getConnection()
          partitionOfRecords.foreach(record => connection.send(record))
          ConnectionPool.returnConnection(connection)  // return to the pool for future reuse
      })
  })

注意：池中的连接应该是根据需要记性懒式创建并且在一段时间之内不再使用需要将其释放，这种方式是最高效的。

其它需要注意的