spark|spark streaming 处理kafka中积压的数据 Kafka数据堆积|sparkStreaming|kafk

spark streaming冷启动处理kafka中积压的数据因为首次启动JOB的时候，由于冷启动会造成内存使用太大，为了防止这种情况出现，限制首次处理的数据量

spark.streaming.backpressure.enabled=true spark.streaming.backpressure.initialRate=200

举个例子：

#!/bin/sh TaskName="funnel" UserName="hadoop" cd `dirname $0` nohup sudo -u ${UserName} /data/bigdata/spark/bin/spark-submit \ --name ${TaskName} \ --class FunnelMain \ --master yarn \ --deploy-mode cluster \ --executor-memory 2G \ --num-executors 3 \ --conf spark.streaming.backpressure.enabled=true \ --conf spark.streaming.backpressure.initialRate=1000 \ --files /data/apps/funnel/app/conf/conf.properties \ /data/apps/funnel/app/target/apphadoop-1-jar-with-dependencies.jar conf.properties >>../log/${TaskName}.log 2>&1 & exit 0

使用SparkStreaming集成kafka时有几个比较重要的参数：

spark.streaming.stopGracefullyOnShutdown （true / false）默认fasle
确保在kill任务时，能够处理完最后一批数据，再关闭程序，不会发生强制kill导致数据处理中断，没处理完的数据丢失
spark.streaming.backpressure.enabled（true / false）默认false
开启后spark自动根据系统负载选择最优消费速率
spark.streaming.backpressure.initialRate （整数）
默认直接读取所有在（2）开启的情况下，限制第一次批处理应该消费的数据，因为程序冷启动队列里面有大量积压，防止第一次全部读取，造成系统阻塞
spark.streaming.kafka.maxRatePerPartition（整数）
默认直接读取所有限制每秒每个消费线程读取每个kafka分区最大的数据量

特别注意：

只有（4）激活的时候，每次消费的最大数据量，就是设置的数据量，如果不足这个数，就有多少读多少，如果超过这个数字，就读取这个数字的设置的值
只有（2）+（4）激活的时候，每次消费读取的数量最大会等于（4）设置的值，最小是spark根据系统负载自动推断的值，消费的数据量会在这两个范围之内变化根据系统情况，但第一次启动会有多少读多少数据。此后按（2）+（4）设置规则运行
（2）+（3）+（4）同时激活的时候，跟上一个消费情况基本一样，但第一次消费会得到限制，因为我们设置第一次消费的频率了。

Spark Streaming 重启后Kafka数据堆积调优问题：
当应用由于各种其它因素需要暂停消费时，下一次再次启动后就会有大量积压消息需要进行处理，此时为了保证应用能够正常处理积压数据，需要进行相关调优。
另外对于某个时刻，某个topic写入量突增时，会导致整个kafka集群进行topic分区的leader切换，而此时Streaming程序也会受到影响。
所以针对以上问题我们进行了如下调优：

spark.streaming.concurrentJobs=10：提高Job并发数，读过源码的话会发现，这个参数其实是指定了一个线程池的核心线程数而已，没有指定时，默认为1。
spark.streaming.kafka.maxRatePerPartition=2000：设置每秒每个分区最大获取日志数，控制处理数据量，保证数据均匀处理。
spark.streaming.kafka.maxRetries=50：获取topic分区leaders及其最新offsets时，调大重试次数。
在应用级别配置重试
spark.yarn.maxAppAttempts=5
spark.yarn.am.attemptFailuresValidityInterval=1h

此处需要【注意】：
spark.yarn.maxAppAttempts值不能超过hadoop集群中yarn.resourcemanager.am.max-attempts的值，原因可参照下面的源码或者官网配置。