总的来说,需要考虑以下两点:
1. 有效地运用集群资源去减少每个批次处理的时间
2. 正确的设置batch size,以使得处理速度能跟上接收速度
一. 为了减少处理时间,主要有以下几个优化点:
1. 接收数据的并行度。
每个InputDStream只创建一个Receiver用于接收数据,如果接收数据是系统的瓶颈,可以创建多个InputDStream。配置不同的InputDStream读取数据源的不同分区。比如原先用一个InputDStream读取Kafka的两个topic的数据,可以拆分成两个InputDStream读取不同的Topic。处理时,把两个InputDStream收到的数据合并成一个。
int numStreams = 5; List<JavaPairDStream<String, String>> kafkaStreams = new ArrayList<>(numStreams); for (int i = 0; i < numStreams; i++) { kafkaStreams.add(KafkaUtils.createStream(...)); } JavaPairDStream<String, String> unifiedStream = streamingContext.union(kafkaStreams.get(0), kafkaStreams.subList(1, kafkaStreams.size())); unifiedStream.print();
2. 阻塞间隔
有一个配置参数:spark.streaming.blockInterval,它的意思是每间隔多少秒后,Spark才把接收到的数据组成数据块存到Spark中。官网推荐的最小值是50ms,默认值200ms。假设实例化InputDStream时设置的Duration(batch interval)为1秒(1000ms),那么任务执行时,总共有 1000 / 200 = 5 个block,每个block将对应一个task。如果task的数量少于每台机器配置的core的数量,则说明资源没有被很好的利用。应减少 spark.streaming.blockInterval 或增加batch interval。
注意:官网并没有说要使得task的数量和core的数量一致。能想到两个的原因:(1)receiver也会占用core (2)driver也会占用core
3. 相对于1中所说的增加receiver,一个可供选择的方案是通过调用 inputStream.repartition(<number of partitions>) 对inputDStream的数据进行repartition。这将使接收到的数据分布到指定数量的其它机器上,以供进一步处理。
4. 处理数据的并行度
对于reduceByKey,
reduceByKeyAndWindow,Join 等shuffle操作,默认的并行度由 spark.default.parallelism 控制。可以在shuffle操作中设置partition的数量来覆盖默认值。
5. 数据序列化
在Spark Streaming中,有两类数据会被序列化:
5.1 输入数据。默认情况下,输入数据会以StorageLevel.MEMORY_AND_DISK_SER_2 的方式存储在 executor 的内存中。Receiver会反序列化接收到数据,然后再把它序列化成Spark的序列化格式。这显然会有花销。
5.2 持久化Streaming操作产生的RDD。某些windows操作会持久化需要进行多次处理的数据到内存中,默认的存储方式是StorageLevel.MEMORY_ONLY_SER。
以上两种情况,使用Kyro序列化机制来减少CPU和内存的花销。
6. 启动Task的开销
如果每秒启动的task较多(比如:50个或更高),那么发送task去workers的开销会较大。通过以下方式减少这个开销:以Standalone模式或 coarse-grained Mesos 模式运行Spark程序。详见: Running on Mesos guide
7. 设置正确的Batch Interval
每个时间间隔提交的job应该能处理完这个时间间隔内收到的数据。可以通过Web UI 去查看批处理的时间是否小于interval
二. 内存调优
Spark Streaming应用程序需要的内存依赖于程序中用到的transformation。比如,你使用了window操作,想要处理最后10分钟的数据,这就要求内存能保存这10分钟的所有数据。虽然默认情况下是,内存存不下后会存到磁盘,但是这样的会比较慢。建议尽量加大内存。
垃圾回收也是需要考虑的一方面。可以考虑以下几点来减少GC的开销:
1. 接收的数据和RDD会默认序列化并持久化。开启Kyro序列化机制可减少内存的使用。 设置spark.rdd.compress为TRUE,可减少CPU时间
2. 清除旧的数据。Spark Streaming默认会做这件事。
3. 使用CMS 垃圾收集器。driver端使用 --driver-java-options
( spark-submit的时候)。executor端, 使用spark.executor.extraJavaOptions 这个配置
4. 尝试以下2点:1. 使用OFF_HEAP 存储级别 2. 增加executor,减少heap size
总的来说,需要记住以下几点:
1. 一个InputDSteam对应一个receiver, receiver运行在executor上,因此会占用一个core。Receivers以轮询的方式分配到executors中。
2. 每隔block interval的时间就会生成一个block,所以每个batch interval的时间会生成 batch interval / block interval 个block,每个block对应一个task。block会被BlockManager分发到不同的executor上
3. 在batchInterval内,在driver端生成一个RDD,在此期间内生成的blocks是这个RDD的partitions。每个partition是一个task。
4. 较大的blockInterval 意味着较大的data block。较大的 spark.locality.wait 将增加 block 在本地处理的机会。寻找这两个值的平衡点,以使较大的block在本地处理。
5. 可以使用 inputDstream.repartition(n) 来替代设置 batchInterval 和 blockInterval 来获取较好的并行度。但这会产生shuffle的花销。
6. 一次只能处理一个Job。所以当有多个 InputDStream 时,需要先Union两个 InputDStream。
7. Receiver暂时没有暂停的方法。因此当job的处理时间大于 batchInterval 时,receiver 的内存使用将会持续增加,最终导致 BlockNotFoundException 。使用spark.streaming.receiver.maxRate 可以限制 receiver的接收速度