欢迎您访问程序员文章站本站旨在为大家提供分享程序员计算机编程知识!
您现在的位置是: 首页

SparkStreaming消费Kafka中的数据 使用zookeeper和MySQL保存偏移量的两种方式

程序员文章站 2024-03-14 13:29:10
...

第一种是使用zookeeper保存偏移量

object KafkaDirectZookeeper {

  def main(args: Array[String]): Unit = {
 
    val group = "DirectAndZk"
    val conf = new SparkConf().setAppName(s"${this.getClass.getName}").setMaster("local[2]")
    val ssc = new StreamingContext(conf, Duration(5000))
    val topic = "ditopic"
    //指定kafka的broker地址(sparkStream的Task直连到kafka的分区上,用更加底层的API消费,效率更高)
    val brokerList = "hadoop01:9092,hadoop02:9092,hadoop03:9092"
    //指定zk的地址,后期更新消费的偏移量时使用(以后可以使用Redis、MySQL来记录偏移量)
    val zkQuorum = "hadoop01:2181,hadoop02:2181,hadoop03:2181"
    //创建 stream 时使用的 topic 名字集合,SparkStreaming可同时消费多个topic
    val topics: Set[String] = Set(topic)
    //创建一个 ZKGroupTopicDirs 对象,其实是指定往zk中写入数据的目录,用于保存偏移量
    val topicDirs = new ZKGroupTopicDirs(group, topic)
   // new ZKGroupTopicDirs()
    //获取 zookeeper 中的路径 "/g001/offsets/wordcount/"
    val zkTopicPath = s"${topicDirs.consumerOffsetDir}"

    //准备kafka的参数
    val kafkaParams = Map(
      "metadata.broker.list" -> brokerList,
      "group.id" -> group,
      //从头开始读取数据
      "auto.offset.reset" -> kafka.api.OffsetRequest.SmallestTimeString
    )

    //zookeeper 的host 和 ip,创建一个 client,用于跟新偏移量量的
    //是zookeeper的客户端,可以从zk中读取偏移量数据,并更新偏移量
    val zkClient = new ZkClient(zkQuorum)

    //查询该路径下是否字节点(默认有字节点为我们自己保存不同 partition 时生成的)
    // /g001/offsets/wordcount/0/10001"
    // /g001/offsets/wordcount/1/30001"
    // /g001/offsets/wordcount/2/10001"
    //zkTopicPath  -> /g001/offsets/wordcount/

    val children = zkClient.countChildren(zkTopicPath)

    var kafkaStream: InputDStream[(String, String)] = null

    //如果 zookeeper 中有保存 offset,我们会利用这个 offset 作为 kafkaStream 的起始位置
    var fromOffsets: Map[TopicAndPartition, Long] = Map()

    //如果保存过 offset
    if (children > 0) {
      for (i <- 0 until children) {
        // /g001/offsets/wordcount/0/10001
        // /g001/offsets/wordcount/0
        val partitionOffset = zkClient.readData[String](s"$zkTopicPath/${i}")
        // wordcount/0
        val tp = TopicAndPartition(topic, i)
        //将不同 partition 对应的 offset 增加到 fromOffsets 中
        // wordcount/0 -> 10001
        fromOffsets += (tp -> partitionOffset.toLong)
      }
      //Key: kafka的key   values: "hello tom hello jerry"
         //这个会将 kafka 的消息进行 transform,最终 kafak 的数据都会变成 (kafka的key, message) 这样的 tuple
      val messageHandler = (mmd: MessageAndMetadata[String, String]) => (mmd.key(), mmd.message())

      //通过KafkaUtils创建直连的DStream(fromOffsets参数的作用是:按照前面计算好了的偏移量继续消费数据)
      //[String, String, StringDecoder, StringDecoder,     (String, String)]
      //  key    value    key的解码方式   value的解码方式
      kafkaStream = KafkaUtils.createDirectStream[String, String, StringDecoder, StringDecoder, (String,  String)](ssc, kafkaParams, fromOffsets, messageHandler)
    } else {
      //如果未保存,根据 kafkaParam 的配置使用最新(largest)或者最旧的(smallest) offset
      kafkaStream = KafkaUtils.createDirectStream[String, String, StringDecoder, StringDecoder](ssc,      kafkaParams, topics)
    }

    //偏移量的范围
     var offsetRanges = Array[OffsetRange]()

    //如果你调用了DStream的Transformation,就不能使用直连方式
      kafkaStream.foreachRDD { kafkaRDD =>
      //只有KafkaRDD可以强转成HasOffsetRanges,并获取到偏移量
      offsetRanges = kafkaRDD.asInstanceOf[HasOffsetRanges].offsetRanges
      //val lines: RDD[String] = kafkaRDD.map(_._2)
      //对RDD进行操作,触发Action
       lines.foreachPartition(partition =>
        partition.foreach(x => {
          println(x)
        })
      )
      for (o <- offsetRanges) {
        //  /g001/offsets/wordcount/0
        val zkPath = s"${topicDirs.consumerOffsetDir}/${o.partition}"
        //将该 partition 的 offset 保存到 zookeeper
        //  /g001/offsets/wordcount/0/20000
        ZkUtils.updatePersistentPath(zkClient, zkPath, o.untilOffset.toString)
      }
    }
    ssc.start()
    ssc.awaitTermination()
  }
}

第二种是通过MySQL保存偏移量
注意:这种方式使用的是scalikejdbc
导入以下依赖

    <dependency>
            <groupId>org.scalikejdbc</groupId>
            <artifactId>scalikejdbc_2.11</artifactId>
            <version>2.5.0</version>
        </dependency>
        <dependency>
            <groupId>org.scalikejdbc</groupId>
            <artifactId>scalikejdbc-core_2.11</artifactId>
            <version>2.5.0</version>
        </dependency>
        <dependency>
            <groupId>org.scalikejdbc</groupId>
            <artifactId>scalikejdbc-config_2.11</artifactId>
            <version>2.5.0</version>
        </dependency>

需要配置以下数据库连接

db.default.driver="com.mysql.jdbc.Driver"
db.default.url="jdbc:mysql://localhost:3306/test?characterEncoding="utf-8""
db.default.user="root"
db.default.password="root"
object SparkStreamingOffsetMysql {
  def main(args: Array[String]): Unit = {
    val conf = new SparkConf().setAppName("ssom").setMaster("local[2]")
    val ssc = new StreamingContext(conf, Seconds(3))
    val groupId = "didi"
    val brokerList = "hadoop01:9092,hadoop02:9092,hadoop03:9092"
    val topic = "ditopic"
    val topics = Set(topic)
    val kafkas = Map(
      "metadata.broker.list" -> brokerList,
      "group.id" -> groupId,
      "auto.offset.reset" -> kafka.api.OffsetRequest.SmallestTimeString)
      DBs.setup()
     // 直接查询mysql中的offset
       val fromOffset: Map[TopicAndPartition, Long] =
       DB.readOnly {
        implicit session => {
          SQL(s"select * from offset where groupId = '${groupId}'")
            //查询出来后 将数据赋值给元组
            .map(m => (TopicAndPartition(
            m.string("topic"), m.int("partitions")), m.long("untilOffset")))
            .toList().apply()
        }.toMap //最后要toMap因为前面的返回值已经给定
      }
     //创建一个InputDStram 然后根据offset读取数据
     var kafkaStream: InputDStream[(String, String)] = null
     //从mysql中获取数据进行判断
     if (fromOffset.size == 0) {
      //如果程序第一次启动
      kafkaStream = KafkaUtils.createDirectStream[String, String, StringDecoder, StringDecoder](
        ssc, kafkas, topics)
    } else {
      //如果程序不是第一次启动
      var checckOffset = Map[TopicAndPartition, Long]()
      val kafkaCluster = new KafkaCluster(kafkas)
      val earliesOffset: Either[Err, Map[TopicAndPartition, KafkaCluster.LeaderOffset]] =
        kafkaCluster.getEarliestLeaderOffsets(fromOffset.keySet)
      //然后开始比较大小 用Mysql中的offset和kafka的offset进行比较
      if (earliesOffset.isRight) {
        val topicAndPartitionOffset: Map[TopicAndPartition, KafkaCluster.LeaderOffset] = earliesOffset.right.get
        //来个直接进行比较大小
        fromOffset.map(owner => {
          //取kafka汇总的offset
          val topicOffset = topicAndPartitionOffset.get(owner._1).get.offset
          if (owner._2 > topicOffset) {
            owner
          } else {
            (owner._1, topicOffset)
          }
        })
      }
      val messageHandler = (mmd: MessageAndMetadata[String, String]) => {
        (mmd.key(), mmd.message())
      }
      kafkaStream = KafkaUtils.createDirectStream[String, String,
        StringDecoder, StringDecoder, (String, String)](
        ssc, kafkas, checckOffset, messageHandler)
    }
    kafkaStream.foreachRDD(kafkaRDD => {
      val offsetRanges = kafkaRDD.asInstanceOf[HasOffsetRanges].offsetRanges
     kafkaRDD.map(_._2).foreachPartition(partition =>
        partition.foreach(x => {
          println(x)
        })
     
      DB.localTx {
        implicit session =>
          for (os <- offsetRanges) {
            /*  SQL("update offset set groupId=?,topic=?,partitions=?,untilOffset=?")
             .bind(groupId,os.topic,os.partition,os.untilOffset).update().apply()*/
            SQL("replace into offset(groupId,topic,partitions,untilOffset) values(?,?,?,?)")
              .bind(groupId, os.topic, os.partition, os.untilOffset).update().apply()
          }
      }
    })
    ssc.start()
    ssc.awaitTermination()
  }
}