| 01 02 03 04 05 06 07 08 09 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 | val topic : String = "topic_name" //消费的 topic 名字 val topics : Set[String] = Set(topic) //创建 stream 时使用的 topic 名字集合 val topicDirs = new ZKGroupTopicDirs( "test_spark_streaming_group" , topic) //创建一个 ZKGroupTopicDirs 对象，对保存 val zkTopicPath = s "${topicDirs.consumerOffsetDir}" 获取 zookeeper 中的路径，这里会变成 /consumers/test _ spark _ streaming _ group/offsets/topic _ name val zkClient = new ZkClient( "10.4.232.77:2181" ) //zookeeper 的host 和 ip，创建一个 client val children = zkClient.countChildren(s "${topicDirs.consumerOffsetDir}" ) //查询该路径下是否字节点（默认有字节点为我们自己保存不同 partition 时生成的） var kafkaStream : InputDStream[(String, String)] = null var fromOffsets : Map[TopicAndPartition, Long] = Map() //如果 zookeeper 中有保存 offset，我们会利用这个 offset 作为 kafkaStream 的起始位置 if (children > 0 ) { //如果保存过 offset，这里更好的做法，还应该和 kafka 上最小的 offset 做对比，不然会报 OutOfRange 的错误 for (i <- 0 untilchildren) { val partitionOffset = zkClient.readData[String](s "${topicDirs.consumerOffsetDir}/${i}" ) val tp = TopicAndPartition(topic, i) fromOffsets + = (tp -> partitionOffset.toLong) //将不同 partition 对应的 offset 增加到 fromOffsets 中 logInfo( "@@@@@@ topic[" + topic + "] partition[" + i + "] offset[" + partitionOffset + "] @@@@@@" ) } val messageHandler = (mmd : MessageAndMetadata[String, String]) = > (mmd.topic, mmd.message()) //这个会将 kafka 的消息进行 transform，最终 kafak 的数据都会变成 (topic_name, message) 这样的 tuple kafkaStream = KafkaUtils.createDirectStream[String, String, StringDecoder, StringDecoder, (String, String)](ssc, kafkaParam, fromOffsets, messageHandler) } else { kafkaStream = KafkaUtils.createDirectStream[String, String, StringDecoder, StringDecoder](ssc, kafkaParam, topics) //如果未保存，根据 kafkaParam 的配置使用最新或者最旧的 offset } var offsetRanges = Array[OffsetRange]() kafkaStream.transform{ rdd = > offsetRanges = rdd.asInstanceOf[HasOffsetRanges].offsetRanges //得到该 rdd 对应 kafka 的消息的 offset rdd }.map(msg = > Utils.msgDecode(msg)).foreachRDD { rdd = > for (o <- offsetRanges) { val zkPath = s "${topicDirs.consumerOffsetDir}/${o.partition}" ZkUtils.updatePersistentPath(zkClient, zkPath, o.fromOffset.toString) //将该 partition 的 offset 保存到 zookeeper logInfo(s "@@@@@@ topic ${o.topic} partition ${o.partition} fromoffset ${o.fromOffset} untiloffset ${o.untilOffset} #######" ) } rdd.foreachPartition( message = > { while (message.hasNext) { logInfo(s "@^_^@ [" + message.next() + "] @^_^@" ) } } ) } |
| 01 02 03 04 05 06 07 08 09 10 11 | val partitionOffset = zkClient.readData[String](s "${topicDirs.consumerOffsetDir}/${i}" )val tp = TopicAndPartition(topic, i) val requestMin = OffsetRequest(Map(tp -> PartitionOffsetRequestInfo(OffsetRequest.EarliestTime, 1 )))val consumerMin = new SimpleConsumer( "broker_host" , 9092 , 10000 , 10000 , "getMinOffset" ) //注意这里的 broker_host，因为这里会导致查询不到，解决方法在下面val curOffsets = consumerMin.getOffsetsBefore(requestMin).partitionErrorAndOffsets(tp).offsetsvar nextOffset = partitionOffset.toLongif (curOffsets.length > 0 && nextOffset < curOffsets.head) { // 通过比较从 kafka 上该 partition 的最小 offset 和 zk 上保存的 offset，进行选择 nextOffset = curOffsets.head}fromOffsets + = (tp -> nextOffset) //设置正确的 offset，这里将 nextOffset 设置为 0（0 只是一个特殊值），可以观察到 offset 过期的想想 |
| 01 02 03 04 05 06 07 08 09 10 11 12 | val topic _ name = "topic_name" //topic_name 表示我们希望获取的 topic 名字val topic 2 = List(topic _ name) val req = new TopicMetadataRequest(topic 2 , 0 )val getLeaderConsumer = new SimpleConsumer( "broker_host" , 9092 , 10000 , 10000 , "OffsetLookup" ) // 第一个参数是 kafka broker 的host，第二个是 portval res = getLeaderConsumer.send(req)val topicMetaOption = res.topicsMetadata.headOptionval partitions = topicMetaOption match { case Some(tm) = > tm.partitionsMetadata.map(pm = > (pm.partitionId, pm.leader.get.host)).toMap[Int, String] // 将结果转化为 partition -> leader 的映射关系 case None = > Map[Int, String]()} |