总的来说,需要考虑以下两点:

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的接收速度

05-11 19:57