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.png2.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
网友评论