1.基础概念

• SparkConext
  SparkConext是Spark 框架的核心，其中包含了DAG 以及 TaskScheduler 的实例，用于Spark任务的构建，资源协调等工作
• StreamingContext
  Spark Streaming的上下文，主要是Streaming 程序的DAG构建、启动入口
• DStreams
  一段时间内的RDD集合，主要是Spark Streaming 的Batch Data的逻辑结构

2.Streaming 的数据流转图

Spark Streaming 的处理逻辑为从数据源（Source）读取数据，进行数据处理（Transform），最终执行数据输出（Sink）

image.png

2.1 Source

• 基础Source（文件系统）
  Spark Streaming 把文件系统作为Source的时候，会监听指定文件夹。根据文件的变更时间作为依据，读取作为Streaming 程序的Batch数据生成依据。其中需要注意一下几点：
  a. 如果在文件一个Streaming Batch内多次变更，仅会读取一次。也就是说多次变更可能会造成数据丢失，所以建议在其他文件目录进行写入，直接移动到监听目录
  b. 针对S3 之类的云存储，由于文件移动次数限制。以及重名是Cpoy数据的逻辑，所以不能采取重命名操作。建议采取其他方案
• 消息处理框架
  Kafaka: 参见官网文档
  Kinesis: 参见官网文档
• 自定义Source
  需要自定义Recievers： 参见官网文档

2.2 Transform

大部分参见官网，这里主要提一下updateStateByKey.
updateStateByKey 主要是基于前面Batch的数据 和 当前Btach数据进行统一聚合计算。示例如下：

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

2.3. Sink

Spark Streaming 的Sink 主要是 基于foreachRDD、foreachPartition 进行数据的输出。切记必须要有action操作，否则基于Spark 是lazy的特性，所有的数据都会丢失。示例如下：

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

3. Checkpointing

Checkpointing 主要包含两部分数据：

• Metadata ：
  主要是包含任务启动的Config 以及 Streaming的 DAG配置。主要用于Driver的重启
• Data :
  主要是RDD的数据缓存，用于Executor的重启。需要注意一点：如果是stateful的Streaming， 考虑到依赖，会定时进行Checkpoint，避免每一个RDD都缓存造成存储压力
• 关于Accumulators, Broadcast Variables的Checkpoint
  Accumulators, Broadcast Variables不会从chenkpoint中进行回复，如果必须要进行恢复的话。可以采取lazy实例化的方式，具体如下：

object WordExcludeList {

  @volatile private var instance: Broadcast[Seq[String]] = null

  def getInstance(sc: SparkContext): Broadcast[Seq[String]] = {
    if (instance == null) {
      synchronized {
        if (instance == null) {
          val wordExcludeList = Seq("a", "b", "c")
          instance = sc.broadcast(wordExcludeList)
        }
      }
    }
    instance
  }
}

object DroppedWordsCounter {

  @volatile private var instance: LongAccumulator = null

  def getInstance(sc: SparkContext): LongAccumulator = {
    if (instance == null) {
      synchronized {
        if (instance == null) {
          instance = sc.longAccumulator("DroppedWordsCounter")
        }
      }
    }
    instance
  }
}

wordCounts.foreachRDD { (rdd: RDD[(String, Int)], time: Time) =>
  // Get or register the excludeList Broadcast
  val excludeList = WordExcludeList.getInstance(rdd.sparkContext)
  // Get or register the droppedWordsCounter Accumulator
  val droppedWordsCounter = DroppedWordsCounter.getInstance(rdd.sparkContext)
  // Use excludeList to drop words and use droppedWordsCounter to count them
  val counts = rdd.filter { case (word, count) =>
    if (excludeList.value.contains(word)) {
      droppedWordsCounter.add(count)
      false
    } else {
      true
    }
  }.collect().mkString("[", ", ", "]")
  val output = "Counts at time " + time + " " + counts
})

4. 调优

• Reciever的速率：主要是考虑数据处理不及时，控制接收数率

spark.streaming.receiver.maxRate = 10

• 设置Reciever读取数据组成Block的时间间隔
  这个参数主要用于控制读取数据任务的并发度，计算公式为：每个批次的时间间隔/生成Block的时间间隔。如果任务数量比CPU数量少太多，建议调整。比如：
     成Block的时间间隔 = 200ms
     每个批次的时间间隔= 2s
     则 生成的读取任务的数量为： 2/200 = 10
  如果想调整生成任务数量，需要设置一下参数：

spark.streaming.blockInterval = 200

• 设置任务任务并行度。通过一下参数：

spark.default.parallelism