Spark Streaming 编程指南
2017-09-01 15:34:41 0 举报AI智能生成
Spark Streaming 编程指南
作者其他创作
大纲/内容
基本概念
代码库依赖
以Maven方式管理依赖
<dependency>
<groupId>org.apache.spark</groupId>
<artifactId>spark-streaming_2.10</artifactId>
<version>1.6.1</version>
</dependency> 以SBT方式管理依赖
初始化StreamingContext
根据conf创建
val ssc = new StreamingContext(conf, Seconds(1)) 根据已有的SparkContext创建
val ssc = new StreamingContext(sc, Seconds(1))离散数据流 (DStreams)
离散数据流（DStream）是Spark Streaming最基本的抽象，代表了一种连续数据流
输入DStream和接收器（二者相关联）
输入DStream
基础数据源
文件数据流（不依赖接收器）
streamingContext.fileStream(dataDirectory)
streamingContext.textFileStream(dataDirectory)
套接字
streamingContext.socketTextStream(host, port) Akka actor
streamingContext.actorStream(actorProps, actor-name)
高级数据源
Kafka
Flume
Kinesis
自定义数据源
只需实现一个接收器（Receiver）
接收器
接收器可靠性
不可靠接收器
不可靠接收器不会发送任何确认信息。不过这种接收器常用于不支持确认的数据源，或者不想引入数据确认的复杂性的数据源
可靠接收器
可靠接收器会在成功接收并保存好Spark数据副本后，向可靠数据源发送确认信息
DStream算子
transformation算子
基于窗口的算子
window(windowLength, slideInterval)
countByWindow(windowLength, slideInterval)
reduceByWindow(func, windowLength, slideInterval)
reduceByKeyAndWindow(func,windowLength, slideInterval, [numTasks])
reduceByKeyAndWindow(func, invFunc,windowLength,slideInterval, [numTasks])
与上一个的区别：会用之前滑动窗口计算结果，递增地计算每个窗口的归约结果
countByValueAndWindow(windowLength,slideInterval, [numTasks])
Join相关的算子
流-流（Stream-Stream）关联
可以调用window函数做基于窗口关联
流-数据集（stream-dataset）关联
不触发shuffle
map(func)
filter(func)
flatMap(func)
union(otherStream)
触发shuffle
repartition(numPartitions)
count()
reduce(func)
countByValue()
recudeByKey(func,[numTasks])
cogroup(otherStream,[numTasks])
transform(func)
transform算子（及其变体transformWith）可以支持任意的RDD到RDD的映射操作
val cleanedDStream = wordCounts.transform(rdd => {
rdd.join(spamInfoRDD).filter(...) // 将DStream中的RDD和spamInfoRDD关联，并实时过滤垃圾数据
...
}) join(otherStream,[numTasks])
updateStateByKey(func)
按key更新状态
DStream输出算子
print()
saveAsTextFiles(prefix, [suffix])
saveAsHadoopFiles(prefix, [suffix])
foreachRDD(func)
DataFrame和SQL相关的算子
MLlib算子
缓存/持久化
一般Dstream持久化
调用persist()
该方法内部会自动调用DStream中每个RDD的persist方法进而将数据持久化到内存中
滑动窗口算子持久化
滑动窗口算子产生的DStream对象默认会自动持久化到内存中（不需要开发者调用persist）
网络接收数据的输入数据流
默认的持久化级别会将数据持久化到两个不同的节点上互为备份副本
检查点
检查点需要保存的两种数据
元数据检查点（恢复驱动器节点上的故障）
Configuration – 创建Streaming应用程序的配置信息
DStream operations – 定义流式处理逻辑的DStream操作信息
Incomplete batches – 已经排队但未处理完的批次信息
数据检查点（恢复有状态转换操作）
将生成的RDD保存到可靠的存储中
启用与配置
何时启用检查点
使用了有状态的转换算子时
需要支持驱动器故障中恢复时
如何配置检查点
启用有状态的转换算子检查点
streamingContext.checkpoint(checkpointDirectory)
驱动器故障后恢复
如果程序是首次启动，就需要new一个新的StreamingContext，并定义好所有的数据流处理，然后调用StreamingContext.start()
如果程序是故障后重启，就需要从检查点目录中的数据中重新构建StreamingContext对象
性能调优
减少批次处理时间三种方式
增加数据接收并发度三种方式
优化接收器阻塞间隔
数据块间隔配置参数spark.streaming.blockInterval
增加数据流接收的并行度
val numStreams = 5
val kafkaStreams = (1 to numStreams).map { i => KafkaUtils.createStream(...) }
val unifiedStream = streamingContext.union(kafkaStreams)
unifiedStream.print() 对输入DStream划分分区
inputStream.repartition(<number of partitions>)
增加数据处理并发度
修改并发任务数参数spark.default.parallelism提高任务并发数量
数据序列化（Kryo序列化）
序列化输入数据
Streaming算子所生产的持久化的RDDs
减少任务启动开销两种方式
使用Kryo来序列化任务
配置参数spark.closure.serializer
执行模式
Spark独立部署或者Mesos粗粒度模式下任务的启动时间比Mesos细粒度模式下的任务启动时间要短
设置合适的批次间隔
批次数据的处理速度应该和其生成速度一样快
内存调优
DStream持久化
Kryo序列化
数据压缩
配置参数spark.rdd.compress
清除老数据
默认情况下，所有的输入数据以及DStream的transformation算子产生的持久化RDD都是自动清理的
CMS垃圾回收器
设置驱动器CMS垃圾回收器（通过spark-submit中的–driver-java-options设置）
设置执行器CMS垃圾回收器（使用spark.executor.extraJavaOptions配置参数）
其他提示
配合Tachyon使用堆外内存来持久化RDD
使用更多但是更小的执行器进程。这样GC压力就会分散到更多的JVM堆中
每个批次中数据块的个数将会决定处理这批数据并行任务的个数
输入数据默认存储级别：StorageLevel.MEMORY_AND_DISK_SER_2
在application中同时设置
容错语义
收藏
职业：暂无
0 条评论
下一页
