SparkStreaming实现HA高可用

比眉伴天荒 2023-11-11 06:48 112阅读 0赞

### SparkStreaming HA高可用性实现起来首先是自身的高可用操作主要是以下三点 ###

#### 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();
    
    }

二、Scala版

package cn.piesat.spark
    
    import org.apache.kafka.clients.consumer.{ConsumerRecord}
    import org.apache.kafka.common.serialization.StringDeserializer
    import org.apache.spark.sql.SparkSession
    import org.apache.spark.streaming.dstream.InputDStream
    import org.apache.spark.streaming.kafka010.{ConsumerStrategies, KafkaUtils, LocationStrategies}
    import org.apache.spark.streaming.{Seconds, StreamingContext}
    
    object SparkStreamingKafka {
      private val brokers = "hadoop01:9092"
      private val topics = "lj01"
      private val checkPointPath = "hdfs://hadoop01:9000/sparkStreaming/kafka6"
    
      def main(args: Array[String]): Unit = {
        val spark = getSparkSession()
        val streamingContext = StreamingContext.getOrCreate(checkPointPath, () => {
          val ssc = new StreamingContext(spark.sparkContext, Seconds(5))
          ssc.checkpoint(checkPointPath)
          val kafkaInputStream = getKafkaInputStream(ssc)
          val result = kafkaInputStream.map(x => x.value()).flatMap(x => {
            x.split(" ").map(x => {
              (x, 1)
            })
          }).reduceByKey(_ + _)
          result.print()
          ssc
        })
        streamingContext.start()
        streamingContext.awaitTermination()
      }
    
      def getSparkSession(): SparkSession = {
        SparkSession.builder()
          .appName("kafka_test")
          .master("local[4]")
          .config("spark.serializer", "org.apache.spark.serializer.KryoSerializer")
          .getOrCreate()
      }
    
      def getKafkaInputStream(ssc: StreamingContext): InputDStream[ConsumerRecord[String, String]] 
    
    = {
        val topicArray = topics.split(",").toList
        val kafkaParams = Map[String, Object](
          "bootstrap.servers" -> brokers,
          "key.deserializer" -> classOf[StringDeserializer],
          "value.deserializer" -> classOf[StringDeserializer],
          "group.id" -> "lj00",
          "auto.offset.reset" -> "latest",
          "enable.auto.commit" -> (false: java.lang.Boolean)
        )
        KafkaUtils.createDirectStream[String, String](
          ssc,
          LocationStrategies.PreferConsistent,
          ConsumerStrategies.Subscribe[String, String](topicArray, kafkaParams)
        )
      }
    
    }

注意：对streaming的操作逻辑必须写在StreamingContext.getOrCreate()方法里，因为若是第二次恢复时则执行方法里的逻辑！！！

#### 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”);