Spark Streaming 之 Kafka 偏移量管理

本文主要介绍 Spark Streaming 应用开发中消费 Kafka 消息的相关内容，文章着重突出了开发环境的配置以及手动管理 Kafka 偏移量的实现。

1、开发环境

一、组件版本

• CDH 集群版本：6.0.1
• Spark 版本：2.2.0
• Kafka 版本：1.0.1

二、Maven 依赖

<!-- scala -->
<dependency>
  <groupId>org.scala-lang</groupId>
  <artifactId>scala-library</artifactId>
  <version>2.11.8</version>
</dependency>

<!-- spark 基础依赖 -->
<dependency>
  <groupId>org.apache.spark</groupId>
  <artifactId>spark-core_2.11</artifactId>
  <version>2.2.0</version>
</dependency>

<!-- spark-streaming 相关依赖 -->
<dependency>
  <groupId>org.apache.spark</groupId>
  <artifactId>spark-streaming_2.11</artifactId>
  <version>2.2.0</version>
</dependency>

<!-- spark-streaming-kafka 相关依赖 -->
<dependency>
  <groupId>org.apache.spark</groupId>
  <artifactId>spark-streaming-kafka-0-10_2.11</artifactId>
  <version>2.2.0</version>
</dependency>

<!-- zookeeper 相关依赖 -->
<dependency>
  <groupId>org.apache.zookeeper</groupId>
  <artifactId>zookeeper</artifactId>
  <version>3.4.5-cdh6.0.1</version>
</dependency>
复制代码

三、scala 编译

在 pom.xml 的 build 节点下的 plugins 中添加 scala 编译插件

<plugin>
    <groupId>org.scala-tools</groupId>
    <artifactId>maven-scala-plugin</artifactId>
    <executions>
      <execution>
        <goals>
          <goal>compile</goal>
          <goal>testCompile</goal>
        </goals>
      </execution>
    </executions>
    <configuration>
      <scalaVersion>${scala.version}</scalaVersion>
      <args>
        <arg>-target:jvm-1.5</arg>
      </args>
    </configuration>
</plugin>
复制代码

Maven 打包语句：mvn clean scala:compile compile package

四、打包注意事项

因为 spark、spark-streaming、zookeeper 等均为大数据集群中必备的组件，所以与之相关的依赖无需打包到最终的 jar 包中，能够将其 scope 设置为 provided 便可；不然最终的 jar 包会至关庞大。

2、Kafka 偏移量

一、偏移量（offset）

这里的偏移量是指 kafka consumer offset，在 Kafka 0.9 版本以前消费者偏移量默认被保存在 zookeeper 中（/consumers/<group.id>/offsets/<topic>/<partitionId>），所以在初始化消费者的时候须要指定 zookeeper.hosts。

随着 Kafka consumer 在实际场景的不断应用，社区发现旧版本 consumer 把位移提交到 ZooKeeper 的作法并不合适。ZooKeeper 本质上只是一个协调服务组件，它并不适合做为位移信息的存储组件，毕竟频繁高并发的读/写操做并非 ZooKeeper 擅长的事情。所以在 0.9 版本开始 consumer 将位移提交到 Kafka 的一个内部 topic（__consumer_offsets）中，该主题默认有 50 个分区，每一个分区 3 个副本。

二、消息交付语义

• at-most-once：最多一次，消息可能丢失，但不会被重复处理；
• at-least-once：至少一次，消息不会丢失，但可能被处理屡次；
• exactly-once：精确一次，消息必定会被处理且只会被处理一次。

若 consumer 在消息消费以前就提交位移，那么即可以实现 at-most-once，由于若 consumer 在提交位移与消息消费之间崩溃，则 consumer 重启后会重新的 offset 位置开始消费，前面的那条消息就丢失了；相反地，若提交位移在消息消费以后，则可实现 at-least-once 语义。因为 Kafka 没有办法保证这两步操做能够在同一个事务中完成，所以 Kafka 默认提供的就是 at-least-once 的处理语义。

三、offset 提交方式

默认状况下，consumer 是自动提交位移的，自动提交间隔是 5 秒，能够经过设置 auto.commit.interval.ms 参数能够控制自动提交的间隔。自动位移提交的优点是下降了用户的开发成本使得用户没必要亲自处理位移提交；劣势是用户不能细粒度地处理位移的提交，特别是在有较强的精确一次处理语义时（在这种状况下，用户可使用手动位移提交）。

所谓的手动位移提交就是用户自行肯定消息什么时候被真正处理完并能够提交位移，用户能够确保只有消息被真正处理完成后再提交位移。若是使用自动位移提交则没法保证这种时序性，所以在这种状况下必须使用手动提交位移。设置使用手动提交位移很是简单，仅仅须要在构建 KafkaConsumer 时设置 enable.auto.commit=false，而后调用 commitSync 或 commitAsync 方法便可。

3、使用 Zookeeper 管理 Kafka 偏移量

一、Zookeeper 管理偏移量的优点

虽说新版 kafka 中已经无需使用 zookeeper 管理偏移量了，可是使用 zookeeper 管理偏移量相比 kafka 自行管理偏移量有以下几点好处：

1. 可使用 zookeeper 管理工具轻松查看 offset 信息；
2. 无需修改 groupId 便可从头读取消息；
3. 特别状况下能够人为修改 offset 信息。

借助 zookeeper 管理工具能够对任何一个节点的信息进行修改、删除，若是但愿从最开始读取消息，则只须要删除 zk 某个节点的数据便可。

二、Zookeeper 偏移量管理实现

import org.I0Itec.zkclient.ZkClient
import org.apache.kafka.clients.consumer.ConsumerRecord
import org.apache.kafka.common.TopicPartition
import org.apache.spark.SparkConf
import org.apache.spark.streaming.StreamingContext
import org.apache.spark.streaming.kafka010.OffsetRange

import scala.collection.JavaConverters._

class ZkKafkaOffset(getClient: () => ZkClient, getZkRoot : () => String) {

  // 定义为 lazy 实现了懒汉式的单例模式，解决了序列化问题，方便使用 broadcast
  lazy val zkClient: ZkClient = getClient()
  lazy val zkRoot: String = getZkRoot()

  // offsetId = md5(groupId+join(topics))
  // 初始化偏移量的 zk 存储路径 zkRoot
  def initOffset(offsetId: String) : Unit = {
    if(!zkClient.exists(zkRoot)){
      zkClient.createPersistent(zkRoot, true)
    }
  }

  // 从 zkRoot 读取偏移量信息
  def getOffset(): Map[TopicPartition, Long] = {
    val keys = zkClient.getChildren(zkRoot)
    var initOffsetMap: Map[TopicPartition, Long] = Map()
    if(!keys.isEmpty){
      for (k:String <- keys.asScala) {
        val ks = k.split("!")
        val value:Long = zkClient.readData(zkRoot + "/" + k)
        initOffsetMap += (new TopicPartition(ks(0), Integer.parseInt(ks(1))) -> value)
      }
    }
    initOffsetMap
  }

  // 根据单条消息，更新偏移量信息
  def updateOffset(consumeRecord: ConsumerRecord[String, String]): Boolean = {
    val path = zkRoot + "/" + consumeRecord.topic + "!" + consumeRecord.partition
    zkClient.writeData(path, consumeRecord.offset())
    true
  }

  // 消费消息前，批量更新偏移量信息
  def updateOffset(offsetRanges: Array[OffsetRange]): Boolean = {
    for (offset: OffsetRange <- offsetRanges) {
      val path = zkRoot + "/" + offset.topic + "!" + offset.partition
      if(!zkClient.exists(path)){
        zkClient.createPersistent(path, offset.fromOffset)
      }
      else{
        zkClient.writeData(path, offset.fromOffset)
      }
    }
    true
  }

  // 消费消息后，批量提交偏移量信息
  def commitOffset(offsetRanges: Array[OffsetRange]): Boolean = {
    for (offset: OffsetRange <- offsetRanges) {
      val path = zkRoot + "/" + offset.topic + "!" + offset.partition
      if(!zkClient.exists(path)){
        zkClient.createPersistent(path, offset.untilOffset)
      }
      else{
        zkClient.writeData(path, offset.untilOffset)
      }
    }
    true
  }

  def finalize(): Unit = {
    zkClient.close()
  }
}

object ZkKafkaOffset{
  def apply(cong: SparkConf, offsetId: String): ZkKafkaOffset = {
    val getClient = () =>{
      val zkHost = cong.get("kafka.zk.hosts", "127.0.0.1:2181")
      new ZkClient(zkHost, 30000)
    }
    val getZkRoot = () =>{
      val zkRoot = "/kafka/ss/offset/" + offsetId
      zkRoot
    }
    new ZkKafkaOffset(getClient, getZkRoot)
  }
}
复制代码

三、Spark Streaming 消费 Kafka 消息

import scala.collection.JavaConverters._

object RtDataLoader {
  def main(args: Array[String]): Unit = {
    // 从配置文件读取 kafka 配置信息
    val props = new Props("xxx.properties")
    val groupId = props.getStr("groupId", "")
    if(StrUtil.isBlank(groupId)){
      StaticLog.error("groupId is empty")
      return
    }
    val kfkServers = props.getStr("kfk_servers")
    if(StrUtil.isBlank(kfkServers)){
      StaticLog.error("bootstrap.servers is empty")
      return
    }
    val topicStr = props.getStr("topics")
    if(StrUtil.isBlank(kfkServers)){
      StaticLog.error("topics is empty")
      return
    }

    // KAFKA 配置设定
    val topics = topicStr.split(",")
    val kafkaConf = Map[String, Object](
      "bootstrap.servers" -> kfkServers,
      "key.deserializer" -> classOf[StringDeserializer],
      "value.deserializer" -> classOf[StringDeserializer],
      "group.id" -> groupId,
      "receive.buffer.bytes" -> (102400: java.lang.Integer),
      "max.partition.fetch.bytes" -> (5252880: java.lang.Integer),
      "auto.offset.reset" -> "earliest",
      "enable.auto.commit" -> (false: java.lang.Boolean)
    )

    val conf = new SparkConf().setAppName("ss-kafka").setIfMissing("spark.master", "local[2]")

    // streaming 相关配置
    conf.set("spark.streaming.stopGracefullyOnShutdown","true")
    conf.set("spark.streaming.backpressure.enabled","true")
    conf.set("spark.streaming.backpressure.initialRate","1000")

    // 设置 zookeeper 链接信息
    conf.set("kafka.zk.hosts", props.getStr("zk_hosts", "sky-01:2181"))

    // 建立 StreamingContext
    val sc = new SparkContext(conf)
    sc.setLogLevel("WARN")
    val ssc = new StreamingContext(sc, Seconds(5))

    // 根据 groupId 和 topics 获取 offset
    val offsetId = SecureUtil.md5(groupId + topics.mkString(","))
    val kafkaOffset = ZkKafkaOffset(ssc.sparkContext.getConf, offsetId)
    kafkaOffset.initOffset(ssc, offsetId)
    val customOffset: Map[TopicPartition, Long] = kafkaOffset.getOffset(ssc)

    // 建立数据流
    var stream:InputDStream[ConsumerRecord[String, String]] = null
    if(topicStr.contains("*")) {
      StaticLog.warn("使用正则匹配读取 kafka 主题：" + topicStr)
      stream = KafkaUtils.createDirectStream[String, String](ssc,
        LocationStrategies.PreferConsistent,
        ConsumerStrategies.SubscribePattern[String, String](Pattern.compile(topicStr), kafkaConf, customOffset))
    }
    else {
      StaticLog.warn("待读取的 kafka 主题：" + topicStr)
      stream = KafkaUtils.createDirectStream[String, String](ssc,
        LocationStrategies.PreferConsistent,
        ConsumerStrategies.Subscribe[String, String](topics, kafkaConf, customOffset))
    }

    // 消费数据
    stream.foreachRDD(rdd => {
      // 消息消费前，更新 offset 信息
      val offsetRanges = rdd.asInstanceOf[HasOffsetRanges].offsetRanges
      kafkaOffset.updateOffset(offsetRanges)
      
      //region 处理详情数据
      StaticLog.info("开始处理 RDD 数据！")
      //endregion
      
      // 消息消费结束，提交 offset 信息
      kafkaOffset.commitOffset(offsetRanges)
    })
    ssc.start()
    ssc.awaitTermination()
  }
}
复制代码

四、注意事项

auto.offset.reset

对于 auto.offset.reset 我的推荐设置为 earliest，初次运行的时候，因为 __consumer_offsets 没有相关偏移量信息，所以消息会从最开始的地方读取；当第二次运行时，因为 __consumer_offsets 已经存在消费的 offset 信息，所以会根据 __consumer_offsets 中记录的偏移信息继续读取数据。

此外，对于使用 zookeeper 管理偏移量而言，只须要删除对应的节点，数据便可从头读取，也是很是方便。不过若是你但愿从最新的地方读取数据，不须要读取旧消息，则能够设置为 latest。

基于正则订阅 Kafka 主题

基于正则订阅主题，有如下好处：

• 无需罗列主题名，一两个主题还好，若是有几十个，罗列过于麻烦了；
• 可实现动态订阅的效果（新增的符合正则的主题也会被读取）。

stream = KafkaUtils.createDirectStream[String, String](ssc,
        LocationStrategies.PreferConsistent,
        ConsumerStrategies.SubscribePattern[String, String](Pattern.compile(topicStr), kafkaConf, customOffset))
复制代码

SparkStreaming 序列化问题

开发 SparkStreaming 程序的每一个人都会遇到各类各样的序列化问题，简单来讲：在 driver 中使用到的变量或者对象无需序列化，传递到 exector 中的变量或者对象须要序列化。所以推荐的作法是，在 exector 中最好只处理数据的转换，在 driver 中对处理的结果进行存储等操做。

stream.foreachRDD(rdd => {
  // driver 代码运行区域
  val offsetRanges = rdd.asInstanceOf[HasOffsetRanges].offsetRanges
  kafkaOffset.updateOffset(offsetRanges)
  
  // exector 代码运行区域
  val resultRDD = rdd.map(xxxxxxxx)
  //endregion
  
  //对结果进行存储
  resultRDD.saveToES(xxxxxx)
  kafkaOffset.commitOffset(offsetRanges)
})
复制代码

文中部分概念摘自《Kafka 实战》，一本很是棒的书籍，推荐一下。

---

Any Code，Code Any！

扫码关注『AnyCode』，编程路上，一块儿前行。