当前位置：首页 > news >正文

SparkStreaming之04：调优

news 2025/8/30 14:20:22

SparkStreaming调优

一、要点

4.1 SparkStreaming运行原理

在这里插入图片描述

深入理解

在这里插入图片描述

4.2 调优策略

4.2.1 调整BlockReceiver的数量

在这里插入图片描述

案例演示：

object MultiReceiverNetworkWordCount {def main(args: Array[String]) {val sparkConf = new SparkConf().setAppName("NetworkWordCount")val sc = new SparkContext(sparkConf)// Create the context with a 1 second batch sizeval ssc = new StreamingContext(sc, Seconds(5))//创建多个接收器(ReceiverInputDStream)，这个接收器接收一台机器上的某个端口通过socket发送过来的数据并处理val lines1 = ssc.socketTextStream("master", 9998, StorageLevel.MEMORY_AND_DISK_SER)val lines2 = ssc.socketTextStream("master", 9997, StorageLevel.MEMORY_AND_DISK_SER)val lines = lines1.union(lines2)lines.repartition(100)//处理的逻辑，就是简单的进行word countval words = lines.repartition(100).flatMap(_.split(" "))val wordCounts = words.map(x => (x, 1)).reduceByKey((a: Int, b: Int) => a + b, new HashPartitioner(10))//将结果输出到控制台wordCounts.print()//启动Streaming处理流ssc.start()//等待Streaming程序终止ssc.awaitTermination()ssc.stop(false)}
}

⭐️4.2.2 调整Block的数量

batchInterval : 触发批处理的时间间隔
blockInterval :将接收到的数据生成Block的时间间隔，spark.streaming.blockInterval(默认是200ms)，那么，BlockRDD的分区数 = batchInterval / blockInterval，即一个Block就是RDD的一个分区，就是一个task
比如，batchInterval是2秒，而blockInterval是200ms，那么task数为10，如果task的数量太少，比一个executor的core数还少的话，那么可以减少blockInterval，blockInterval最好不要小于50ms，太小的话导致task数太多，那么launch task的时间久多了

4.2.3 调整Receiver的接受速率

pps:permits per second 每秒允许接受的数据量(QPS -> queries per second)
Spark Streaming默认的PPS是没有限制的,可以通过参数spark.streaming.receiver.maxRate来控制，默认是Long.Maxvalue

⭐️4.2.3 调整数据处理的并行度

BlockRDD的分区数

a. 通过Receiver接受数据的特点决定

b. 也可以自己通过repartition设置

ShuffleRDD的分区数

a. 默认的分区数为spark.default.parallelism(core的大小)

b. 通过我们自己设置决定

val wordCounts = words.map(x => (x, 1)).reduceByKey((a: Int, b: Int) => a + b, new HashPartitioner(10))

4.2.4 数据的序列化

SparkStreaming两种需要序列化的数据：
a. 输入的数据：默认是以StorageLevel.MEMORY_AND_DISK_SER_2的形式存储在executor上的内存中
b. 缓存的数据：默认是以StorageLevel.MEMORY_ONLY_SER的形式存储的内存中
使用Kryo序列化机制，比Java序列化机制性能好

val conf = new SparkConf().setMaster(...).setAppName(...)
conf.set("spark.serializer", "org.apache.spark.serializer.KryoSerializer")
conf.registerKryoClasses(Array(classOf[MyClass1], classOf[MyClass2]))
val sc = new SparkContext(conf)

4.2.5 内存调优

（1）需要内存大小

和transformation的类型有关，如果使用的是updateStateByKey，Window这样的算子，那么内存就要设置得偏大

（2）数据存储级别

如果把接收到的数据设置的存储级别是MEMORY_DISK这种级别，也就是说如果内存不够可以把数据存储到磁盘上，其实性能还是不好的，性能最好的就是所有的数据都在内存里面，所以如果在资源允许的情况下，把内存调大一点，让所有的数据都存在内存里面。

4.2.6 Outout性能

（1）MySQL，HBase

在这里插入图片描述

（2）Kafka（0.8版本）

虽然现在的Kafka的版本已经到2.x版本了，但是很多公司因为历史遗留的原因，公司里面还是会有0.8x的Kafka。比如本人公司里面有两个Kafka集群，一个是0.8x的kafka，一个是1.x的Kafka。开发的时候有时候需要我们使用SparkStreaming做实时的ETL，然后再把数据打回Kafka，0.8版本的kafka默认是没有批量提交的功能的。本人公司里面一个真实的案例，一位同学写的SparkStreaming程序将数据处理完了以后通过ForeachRDD把数据写回到0.8Kafka。但是数据处理得很慢，经常会收到延时告警。最终发现他把数据写到Kafka的时候是一条数据一条数据提交的性能很差。最终手动实现了批量提交的功能。从此再也没有收到过告警。