文档章节

Spark自定义维护kafka的offset到zk

hblt-j
 hblt-j
发布于 11/15 16:47
字数 387
阅读 8
收藏 0
import kafka.common.TopicAndPartition
import kafka.message.MessageAndMetadata
import kafka.serializer.StringDecoder
import kafka.utils.ZkUtils
import org.I0Itec.zkclient.ZkClient
import org.apache.spark.SparkConf
import org.apache.spark.rdd.RDD
import org.apache.spark.streaming.dstream.InputDStream
import org.apache.spark.streaming.kafka.{HasOffsetRanges, KafkaUtils}
import org.apache.spark.streaming.{Seconds, StreamingContext}

object DirectKafkaExample {

  def main(args: Array[String]) {

    val ssc =  setupSsc
    ssc.start()
    ssc.awaitTermination()


  }


  def setupSsc(): StreamingContext ={

    val conf = new SparkConf().setAppName("CustomDirectKafkaExample").setMaster("local")
    val kafkaParams:Map[String,String] = Map("metadata.broker.list" -> "slave1:9092,slave2:9092,slave3:9092")
    val topicsSet = Set("testha")
    val ssc = new StreamingContext(conf, Seconds(5))

    val messages = createCustomDirectKafkaStream(ssc,kafkaParams,"master0:2181,slave1:2181,slave3:2181","/mysefloffset", topicsSet).map(_._2)

    messages.foreachRDD{rdd => {
      rdd.foreachPartition { partitionOfRecords =>
        if(partitionOfRecords.isEmpty)
          {
            println("此分区数据为空.")
          }
          else
          {
            partitionOfRecords.foreach(println(_))
          }
      }

     }
    }
    ssc
  }


  def createCustomDirectKafkaStream(ssc: StreamingContext, kafkaParams: Map[String, String], zkHosts: String
                                    , zkPath: String, topics: Set[String]): InputDStream[(String, String)] = {
    val topic = topics.last
    val zkClient = new ZkClient(zkHosts, 30000, 30000)
    val storedOffsets = readOffsets(zkClient,zkHosts, zkPath, topic)

    val kafkaStream = storedOffsets match {
          case None => //最新的offset
            KafkaUtils.createDirectStream[String, String, StringDecoder, StringDecoder](ssc, kafkaParams, topics)

          case Some(fromOffsets) => // offset从上次继续开始
            val messageHandler = (mmd: MessageAndMetadata[String, String]) => (mmd.key, mmd.message)
            KafkaUtils.createDirectStream[String, String, StringDecoder, StringDecoder,(String, String)](ssc, kafkaParams, fromOffsets, messageHandler)
        }

    // save the offsets
    kafkaStream.foreachRDD(rdd => saveOffsets(zkClient,zkHosts, zkPath, rdd))
    kafkaStream

  }



  private def readOffsets(zkClient: ZkClient,zkHosts:String, zkPath: String, topic: String):Option[Map[TopicAndPartition, Long]] = {

     println("开始读取从zk中读取offset")

    val stopwatch = new Stopwatch()

    val (offsetsRangesStrOpt, _) = ZkUtils.readDataMaybeNull(zkClient, zkPath)
    offsetsRangesStrOpt match {
      case Some(offsetsRangesStr) =>
          println(s"读取到的offset范围: ${offsetsRangesStr}")
        val offsets = offsetsRangesStr.split(",")
          .map(s => s.split(":"))
          .map { case Array(partitionStr, offsetStr) => (TopicAndPartition(topic, partitionStr.toInt) -> offsetStr.toLong) }
          .toMap
          println("读取offset结束: " + stopwatch)
        Some(offsets)
      case None =>
          println("读取offset结束: " + stopwatch)
        None
    }
  }

  private def saveOffsets(zkClient: ZkClient,zkHosts:String, zkPath: String, rdd: RDD[_]): Unit = {
    println("开始保存offset到zk中去")

    val stopwatch = new Stopwatch()
    val offsetsRanges = rdd.asInstanceOf[HasOffsetRanges].offsetRanges

    //分区,offset
    offsetsRanges.foreach(offsetRange => println(s"Using ${offsetRange}"))

    val offsetsRangesStr = offsetsRanges.map(offsetRange => s"${offsetRange.partition}:${offsetRange.fromOffset}").mkString(",")
     println("保存的偏移量范围:"+ offsetsRangesStr)
    ZkUtils.updatePersistentPath(zkClient, zkPath, offsetsRangesStr)
    println("保存结束,耗时 :" + stopwatch)
  }

  class Stopwatch {
    private val start = System.currentTimeMillis()
    override def toString() = (System.currentTimeMillis() - start) + " ms"
  }


}

本文转载自:http://www.bubuko.com/infodetail-2352528.html

共有 人打赏支持
hblt-j
粉丝 19
博文 148
码字总数 63579
作品 0
海淀
架构师
私信 提问
Spark Streaming 中使用kafka低级api+zookeeper 保存 offset 并重用 以及 相关代码整合

在 Spark Streaming 中消费 Kafka 数据的时候,有两种方式分别是 1)基于 Receiver-based 的 createStream 方法和 2)Direct Approach (No Receivers) 方式的 createDirectStream 方法,详细...

cql252283126
04/13
0
0
Spark Streaming整合kafka实战

kafka作为一个实时的分布式消息队列,实时的生产和消费消息,这里我们可以利用SparkStreaming实时计算框架实时地读取kafka中的数据然后进行计算。在spark1.3版本后,kafkaUtils里面提供了两个...

hblt-j
11/15
0
0
Spark-Streaming获取kafka数据的两种方式-Receiver与Direct的方式

Spark-Streaming获取kafka数据的两种方式-Receiver与Direct的方式,可以从代码中简单理解成Receiver方式是通过zookeeper来连接kafka队列,Direct方式是直接连接到kafka的节点上获取数据了。 ...

张欢19933
2017/06/19
0
0
spark-streaming-kafka之createDirectStream模式

最近一直在用directstream方式消费kafka中的数据,特此总结,整个代码工程分为三个部分 一. 完整工程代码如下(某些地方特意做了说明, 这个代码的部分函数直接用的是spark-streaming-kafka-0...

hblt-j
11/16
0
0
Spark Streaming管理Kafka偏移量

前言 为了让Spark Streaming消费kafka的数据不丢数据,可以创建Kafka Direct DStream,由Spark Streaming自己管理offset,并不是存到zookeeper。启用Spark Streaming的 checkpoints是存储偏移...

架构师springboot
昨天
0
0

没有更多内容

加载失败,请刷新页面

加载更多

Java 11 已发布,String 还能这样玩!

在文章《Java 11 正式发布,这 8 个逆天新特性教你写出更牛逼的代码》中,我有介绍到 Java 11 的八个新特性,其中关于 String 加强部分,我觉得有点意思,这里单独再拉出来讲。 Java 11 增加...

Java技术栈
14分钟前
0
0
什么是以太坊DAO?(二)

Decentralized Autonomous Organization,简称DAO,以太坊中重要的概念。一般翻译为去中心化的自治组织。 在上一节中,我们为了展示什么是DAO创建了一个合约,就像一个采用邀请制的俱乐部,会...

geek12345
29分钟前
4
0
全屋WiFi彻底无死角 这才是终极解决方案

无线网络现在不仅在家庭中不可或缺,在酒店、医院、学校等场景中的需求也越来越多。尤其是这些场景中,房间多但也需要每个房间都能够完美覆盖WiFi,传统的吸顶式AP就无法很好的解决问题。 H3...

linux-tao
43分钟前
4
0
Python日期字符串比较

需要用python的脚本来快速检测一个文件内的二个时间日期字符串的大小,其实实现很简单,首先一些基础的日期格式化知识如下 复制代码 %a星期的简写。如 星期三为Web %A星期的全写。如 星期三为...

dragon_tech
43分钟前
3
0
ORA 各种oraclesql错误

ORA-00001: 违反唯一约束条件 (.) ORA-00017: 请求会话以设置跟踪事件 ORA-00018: 超出最大会话数 ORA-00019: 超出最大会话许可数 ORA-00020: 超出最大进程数 () ORA-00021: 会话附属于其它某...

青峰Jun19er
47分钟前
3
0

没有更多内容

加载失败,请刷新页面

加载更多

返回顶部
顶部