1、保证元数据恢复，就是Driver端挂了之后数据仍然可以恢复
// 创建StreamingContext对象 val ssc: StreamingContext = StreamingContext.getOrCreate(checkpointPath, () => BatchProcessTopic.createContext(brokers, topics, batchseconds, checkpointPath, offset, maxRatePerPartition,jobName,localFlag,processDay,source)) // BatchProcessTopic对象中调用checkpoint ssc.checkpoint(checkpointDirectory)2、Kafka的调优
//sparkConf初始化 val sparkConf = new SparkConf().setAppName(jobName) .set("spark.default.parallelism", REDUCE_PARALLELISM) //调整程序处理的并行度 .set("spark.streaming.backpressure.enabled", "true") //开启后spark自动根据系统负载选择最优消费速率(反压机制),测试重跑跑数据代码 true->false .set("spark.streaming.kafka.maxRatePerPartition", maxRatePerPartition) //设置每秒每个分区最大获取日志数，控制处理数据量，保证数据均匀处理。 .set("spark.streaming.kafka.maxRetries", KAFKA_RETRIES) //获取topic分区leaders(kafka中leade有时会发生变换)及其最新offsets时，调大重试次数。 .set("spark.streaming.stopGracefullyOnShutdown", "true") //确保在kill任务时，能够处理完最后一批数据，再关闭程序，不会发生强制kill导致数据处理中断，没处理完的数据丢失 .set("spark.streaming.backpressure.initialRate", "10") //限制第一次批处理应该消费的数据，因为程序冷启动 队列里面有大量积压，防止第一次全部读取，造成系统阻塞
