SparkStreaming HA高可用性

Posted 2021-01-07 runnerjack

1、UpdateStateByKey、windows等有状态的操作时，自动进行checkpoint,必须设置checkpoint目录，数据保留一份在容错的文件系统中，一旦内存中的数据丢失，可以从文件系统中读取数据，不需要重新计算。

SparkStreaming.checkpoint("hdfs://ip:port/checkpoint")

 

2、Driver高可用性（Java版）

第一次在创建和启动StreamingContext的时候，那么将持续不断的产生实时计算的元数据并写入检查点，如果driver节点挂掉，那么可以让Spark集群自动重启集群（必须使用yarn cluster模式，spark-submit --deploy-mode cluster --supervise ....），然后继续运行计算程序，没有数据丢失。

private static void testDriverHA() {

　　final Streaming checkpointDir="hdfs://ip:port/checkpoint";

　　JavaStreamingContextFactory contextFactory = new JavaStreamingContextFactory() {

　　@Override
　　public JavaStreamingContext create() {
　　　　SparkConf conf = new SparkConf()
　　　　　　.setMaster("local[2]")
　　　　　　.setAppName("AdClickRealTimeStatSpark");

　　　　JavaStreamingContext jssc = new JavaStreamingContext(
　　　　　　　　　　conf, Durations.seconds(5));
　　　　jssc.checkpoint(checkpointDir);

　　　　Map<String, String> kafkaParams = new HashMap<String, String>();
　　　　kafkaParams.put(Constants.KAFKA_METADATA_BROKER_LIST,
　　　　　　ConfigurationManager.getProperty(Constants.KAFKA_METADATA_BROKER_LIST));
　　　　String kafkaTopics = ConfigurationManager.getProperty(Constants.KAFKA_TOPICS);
　　　　String[] kafkaTopicsSplited = kafkaTopics.split(",");
　　　　Set<String> topics = new HashSet<String>();
　　　　for(String kafkaTopic : kafkaTopicsSplited) {
　　　　　　topics.add(kafkaTopic);
　　　　}

　　　　JavaPairInputDStream<String, String> adRealTimeLogDStream = KafkaUtils.createDirectStream(
　　　　　　jssc,
　　　　　　String.class,
　　　　　　String.class,
　　　　　　StringDecoder.class,
　　　　　　StringDecoder.class,
　　　　　　kafkaParams,
　　　　　　topics);

　　　　JavaPairDStream<String, String> filteredAdRealTimeLogDStream =
　　　　　　filterByBlacklist(adRealTimeLogDStream);
　　　　generateDynamicBlacklist(filteredAdRealTimeLogDStream);
　　　　JavaPairDStream<String, Long> adRealTimeStatDStream = calculateRealTimeStat(
　　　　　　filteredAdRealTimeLogDStream);
　　　　calculateProvinceTop3Ad(adRealTimeStatDStream);
　　　　calculateAdClickCountByWindow(adRealTimeLogDStream);
　　　　return jssc;
　　　　}
　　};

　　JavaStreamingContext context = JavaStreamingContext.getOrCreate(
　　checkpointDir, contextFactory);
　　context.start();
　　context.awaitTermination();

}

 

3、实现RDD高可用性，启动WAL预写日志机制

sparkStreaming从原理上说，是通过receiver来进行数据接收的，接收到时的数据，会被划分成一个个的block，block会被组合成batch，针对一个batch，会创建一个Rdd，启动一个job来执行定义的算子操作。receiver主要接收到数据，那么就会立即将数据写入一份到时容错文件系统（比如hdfs）上的checkpoint目录中的，一份磁盘文件中去，作为数据的冗余副本。

　　SparkConf conf = new SparkConf()
　　　　.setMaster("local[2]")
　　　　.setAppName("AdClickRealTimeStatSpark")
　　　　.set("spark.streaming.receiver.writeAheadLog.enable","true");