使用SparkStreaming获取Kafka中的流式数据并指定手动提交offset

水深无声 2023-01-12 10:17 83阅读 0赞

## 概述 ##

`本篇文章主要有三个示例代码，第一个是基础版使用SparkStreaming读取kafka中的流式数据，但是此种方式使用的是自动提交offset的方式，可能会出现offset已提交，但是数据处理过程中出错，导致数据丢失的情况，所以进行了改进，当数据处理完毕后使用手动提交offset的方法。`

`第二个代码示例是使用指定checkpoint的方式保存offset，此种方式代码会有点复杂，而且有个大问题，会生成很多小文件，需要自己在编写程序去定期处理，所以也不是很建议。第三个代码示例是调用官方api将offset保存到_consumer_offset主题中，所以建议读者直接使用此种方式。此外还可以将offset保存到指定位置，例如musql，redis。`

`题外话：sparkstreaming毕竟不是真正的实时计算，而是原理是将数据流离散化为一个一个的微分段，在精准一致的消费流式数据上远没有flink方便！！！所以sparkstreamign只建议使用在对实时性要求在秒级查询得场景！！！`

## 一，连接kafak获取数据 ##

### 1.1 创建kafkaStream的方法 ###

def createDirectStream[K, V](
          ssc: StreamingContext,   //应用上下文
          locationStrategy:  LocationStrategy,
          consumerStrategy: ConsumerStrategies[K, V]
      ): InputDStream[ConsumerRecord]

### 1.2 locationStrategy:kafka分区消费的策略 ###

LocationStrategies.PreferConsistent：
           大部分情况下都是传入此方法,适用于YARN集群和Kafka集群各自独立的情况！
           尽量在所有的executor上负载均衡的分配Kafka的分区！
           
     LocationStrategies.PreferBrokers:
           仅仅在kafka和Yarn使用同一集群的情况下！每个节点都有NM，Broker时使用！
           （但是NM和Broker都是大量消耗磁盘IO的进程，不建议安装在同一台服务器上）
           
      LocationStrategies.PreferFixed:
           在负载不均衡时，自己指定那些host的executor消费那些特定的分区！！！

### 1.3 consumerStrategy: kafka中数据的消费策略 ###

大部分情况下传入   ConsumerStrategies.Subscribe
     
    kafka中的消费策略：
        独立消费者：不依赖于kafak集群！！！
                   消费的offset不需要借助kafka集群进行存储，自动存储到consumer_offsets主题中
                   不需要kafak集群分配分区，直接在线程中指定，要消费那个主题的那个分区！！！
                   
        非独立消费者：依赖于kafka集群！！！
                   消费的offset，需要借助于kafka集群自动存储到consumer_offsets
                   需要kafka集群为一个消费者组中的多个消费者线程分配分区

### 1.4 代码 ###

import org.apache.kafka.clients.consumer.ConsumerRecord
    import org.apache.spark.streaming.dstream.{ DStream, InputDStream}
    import org.apache.spark.streaming.{ Seconds, StreamingContext}
    import org.apache.spark.streaming.kafka010.{ ConsumerStrategies, KafkaUtils, LocationStrategies}
    
    object StreamingKafak { 
      
      def main(args: Array[String]): Unit = { 
      
        //每三秒采集一次数据
        val streaming: StreamingContext = new StreamingContext("local[*]","kafak_test",Seconds(3))
        
        //设置采集kafka数据的相关参数
        val kafkaParams:Map[String,String]=Map[String,String](
          "group.id"->"testKafka",
          "bootstrap.servers"->"hadoop102:9092,hadoop103:9092",
          "enable.auto.commit"->"true",
          "auto.commit.interval.ms"->"500",
          "auto.offset.reset"->"earliest",
          "client.id"->"client1",
          "key.deserializer" -> "org.apache.kafka.common.serialization.StringDeserializer",
          "value.deserializer" -> "org.apache.kafka.common.serialization.StringDeserializer"
        )
      
        //从kafka中消费数据，获取DStream
        val ds: InputDStream[ConsumerRecord[String, String]] = KafkaUtils.createDirectStream(
          streaming,
          LocationStrategies.PreferConsistent,
          ConsumerStrategies.Subscribe[String, String](List("spark"), kafkaParams)
        )
        
       //获取所有value的集合
        val result1 = ds.map(_.value())
      
        result1.print(1000)
       
        
        //启动应用
        streaming.start()
        
        //一直阻塞进程，直到手动调用stop或者异常终止
        streaming.awaitTermination()
      
      } 
    }

## 1.5 pom文件依赖 ##

<dependencies>
            <dependency>
                <groupId>org.apache.spark</groupId>
                <artifactId>spark-core_2.12</artifactId>
                <version>3.0.0</version>
            </dependency>
            <dependency>
                <groupId>org.apache.spark</groupId>
                <artifactId>spark-streaming_2.12</artifactId>
                <version>3.0.0</version>
            </dependency>
    
            <dependency>
                <groupId>org.apache.spark</groupId>
                <artifactId>spark-streaming-kafka-0-10_2.12</artifactId>
                <version>3.0.0</version>
            </dependency>
    
            <dependency>
                <groupId>com.fasterxml.jackson.core</groupId>
                <artifactId>jackson-core</artifactId>
                <version>2.10.1</version>
            </dependency>
        </dependencies>

## 二、使用Checkpoint方式手动提交offset ##

基础版中kafka消费的的offset是自动提交的，可能出现已经提交了offset，但是消费到的数据在没有及时处理完成时，出现程序崩溃的情况！！！会造成数据丢失！！！

### 2.1 解决方案： ###

> **不能在数据处理完成之前提交offset**！只能解决**不丢数据（at least once）**，不能保证 **精准一致性消费（exectly once）**！！ 如果需要实现exactly once，需要在at least once的基础上借助 **幂等操作|事务** 实现

> **保证at least once核心在于不能在数据处理完成之前提交offset！！！  
> 取消offset的自动提交，改为手动提交！！！**

### 2.2 自己维护offset ###

根据存储位置的不同，分为以下三类：

1.  将offset存储到文件系统中：spark提供了checkpoint机制！！！
    
        调用checkpoint，会检查ck目录是否设置，之后会创建ReliableRDDCheckpointData，顺便将Driver的状态也保存到文件
         系统，发生故障时Driver通过读取之前保存的状态来恢复，重建状态，继续运行！  (如果一个saprkstreaming消费kafka程序，
         状态将含有offset)
         
         
          checkpoint的作用：
                 a) 将RDD中数据，持久化到文件系统
                 b） 用于故障恢复！
                            
          采用checkpoint的弊端：每个采集周期都会向ck目录保存状态，因此会生成大量小文件！！！
2.  将offset提交到\_consumer\_offset主题中，需要调用官方的api
3.  将offset维护到任意位置，例如mysql，redis

SparkStreaming获取kafka数据官网配置：[ http://spark.apache.org/docs/latest/streaming-kafka-0-10-integration.html  
][http_spark.apache.org_docs_latest_streaming-kafka-0-10-integration.html]

### 2.3 代码 ###

package com.saprkstreaming.kafka
    
    import org.apache.spark.streaming.kafka010.{ ConsumerStrategies, KafkaUtils, LocationStrategies}
    import org.apache.spark.streaming.{ Seconds, StreamingContext}
    
    object KafkaStreamingExection { 
      
      def main(args: Array[String]): Unit = { 
        
        //要求在重建StreamingContext的函数中，必须包含重建的所有逻辑
        def createStreamingContext()= { 
      
          //每三秒采集一批数据
          val context = new StreamingContext("local[*]", "testPark", Seconds(3))
      
          //设置ck目录
          context.checkpoint("ck")
          
          //设置kafak连接属性
          val kafkaParam= Map[String, String](
            "group.id" -> "testKafka",
            "bootstrap.servers" -> "hadoop102:9092,hadoop103:9092",
            "enable.auto.commit" -> "false",
            "auto.commit.interval.ms" -> "500",
            "auto.offset.reset" -> "earliest",
            "client.id" -> "client1",
            "key.deserializer" -> "org.apache.kafka.common.serialization.StringDeserializer",
            "value.deserializer" -> "org.apache.kafka.common.serialization.StringDeserializer"
          )
      
          //从kafka中获取DStream
          val ds = KafkaUtils.createDirectStream(
            context,
            LocationStrategies.PreferConsistent,
            ConsumerStrategies.Subscribe[String, String](List("spark"), kafkaParam)
          )
      
          //获取所有的value集合
          val ds1= ds.map(record => { 
            val value: String = record.value()
            //模拟故障，抛出异常
            if (value.equals("d")) { 
              //模拟故障
              //throw new RuntimeException("异常")
              //throw new UnknownError("错误")
            }
            value
          })
          
          //程序的处理逻辑
          ds1.print(1000)
          
          context
        }
        
        //创建StreamingContext环境上下文
        val streamContext = StreamingContext.getActiveOrCreate("ck",createStreamingContext)
        
        //应用启动
        streamContext.start()
        
        //一直阻塞当前进程，直到手动调用stop或者由于异常终止
        streamContext.awaitTermination() 
      } 
    }

### 三、调用官方api手动提交offset保存到\_consumer\_offset主题中 ###

package com.saprkstreaming.kafka
    
    import org.apache.kafka.clients.consumer.ConsumerRecord
    import org.apache.spark.streaming.dstream.InputDStream
    import org.apache.spark.streaming.kafka010.{ CanCommitOffsets, ConsumerStrategies, HasOffsetRanges, KafkaUtils, LocationStrategies, OffsetRange}
    import org.apache.spark.streaming.{ Seconds, StreamingContext}
    
    object KafkaStreamingExection2 { 
      
      def main(args: Array[String]): Unit = { 
        
       //每三秒采集一批数据
        val context: StreamingContext = new StreamingContext("local[*]","testkafka",Seconds(3))
      
        //配置kafka连接属性
        val kafkaParams: Map[String, String] = Map[String, String](
          "group.id" -> "testKafka",
          "bootstrap.servers" -> "hadoop102:9092,hadoop103:9092",
          "enable.auto.commit" -> "false",
          "auto.offset.reset" -> "earliest",
          "client.id" -> "client1",
          "key.deserializer" -> "org.apache.kafka.common.serialization.StringDeserializer",
          "value.deserializer" -> "org.apache.kafka.common.serialization.StringDeserializer"
        )
      
        //从kafka中获取DStream
        val ds: InputDStream[ConsumerRecord[String, String]] = KafkaUtils.createDirectStream(
          context,
          LocationStrategies.PreferConsistent,
          ConsumerStrategies.Subscribe[String, String](List("spark"), kafkaParams)
        )
            
        ds.foreachRDD(rdd=>{ 
          //获取此批数据的offset的范围
          val offsetRange: Array[OffsetRange] = rdd.asInstanceOf[HasOffsetRanges].offsetRanges
          
          //进行业务处理
          rdd.foreach(record=>{ 
            val value: String = record.value()
            if (value.equals("d")){ 
              //模拟故障，抛出异常
              //throw new RuntimeException("异常！！！")
            }
              println(value)
          })
          
          //业务处理完成后，手动提交offset！
          ds.asInstanceOf[CanCommitOffsets].commitAsync(offsetRange)  
        })
            
        //启动应用
        context.start()
        
        //阻塞当前进程，直到手动调用stop或者由于异常终止
        context.awaitTermination() 
      } 
    }

[http_spark.apache.org_docs_latest_streaming-kafka-0-10-integration.html]: http://spark.apache.org/docs/latest/streaming-kafka-0-10-integration.html