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
