您好,登錄后才能下訂單哦!
這篇文章主要介紹“spark streaming窗口聚合操作后怎么管理offset”,在日常操作中,相信很多人在spark streaming窗口聚合操作后怎么管理offset問(wèn)題上存在疑惑,小編查閱了各式資料,整理出簡(jiǎn)單好用的操作方法,希望對(duì)大家解答”spark streaming窗口聚合操作后怎么管理offset”的疑惑有所幫助!接下來(lái),請(qǐng)跟著小編一起來(lái)學(xué)習(xí)吧!
對(duì)于spark streaming來(lái)說(shuō)窗口操作之后,是無(wú)法管理offset的,因?yàn)閛ffset的存儲(chǔ)于HasOffsetRanges。只有kafkaRDD繼承了他,所以假如我們對(duì)KafkaRDD進(jìn)行了轉(zhuǎn)化之后就無(wú)法再獲取offset了。
還有窗口之后的offset的管理,也是很麻煩的,主要原因就是窗口操作會(huì)包含若干批次的RDD數(shù)據(jù),那么提交offset我們只需要提交最近的那個(gè)批次的kafkaRDD的offset即可。如何獲取呢?
對(duì)于spark 來(lái)說(shuō)代碼執(zhí)行位置分為driver和executor,我們希望再driver端獲取到offset,在處理完結(jié)果提交offset,或者直接與結(jié)果一起管理offset。
說(shuō)到driver端執(zhí)行,其實(shí)我們只需要使用transform獲取到offset信息,然后在輸出操作foreachrdd里面使用提交即可。
package bigdata.spark.SparkStreaming.kafka010
import java.util.Properties
import org.apache.kafka.clients.consumer.{Consumer, ConsumerRecord, KafkaConsumer}
import org.apache.kafka.common.TopicPartition
import org.apache.kafka.common.serialization.StringDeserializer
import org.apache.spark.rdd.RDD
import org.apache.spark.streaming.kafka010._
import org.apache.spark.streaming.{Seconds, StreamingContext}
import org.apache.spark.{SparkConf, TaskContext}
import scala.collection.JavaConverters._
import scala.collection.mutable
object kafka010NamedRDD {
def main(args: Array[String]) {
// 創(chuàng)建一個(gè)批處理時(shí)間是2s的context 要增加環(huán)境變量
val sparkConf = new SparkConf().setAppName("DirectKafkaWordCount").setMaster("local[*]")
val ssc = new StreamingContext(sparkConf, Seconds(5))
ssc.checkpoint("/opt/checkpoint")
// 使用broker和topic創(chuàng)建DirectStream
val topicsSet = "test".split(",").toSet
val kafkaParams = Map[String, Object]("bootstrap.servers" -> "mt-mdh.local:9093",
"key.deserializer"->classOf[StringDeserializer],
"value.deserializer"-> classOf[StringDeserializer],
"group.id"->"test4",
"auto.offset.reset" -> "latest",
"enable.auto.commit"->(false: java.lang.Boolean))
// 沒(méi)有接口提供 offset
val messages = KafkaUtils.createDirectStream[String, String](
ssc,
LocationStrategies.PreferConsistent,
ConsumerStrategies.Subscribe[String, String](topicsSet, kafkaParams,getLastOffsets(kafkaParams ,topicsSet)))//
var A:mutable.HashMap[String,Array[OffsetRange]] = new mutable.HashMap()
val trans = messages.transform(r =>{
val offsetRanges = r.asInstanceOf[HasOffsetRanges].offsetRanges
A += ("rdd1"->offsetRanges)
r
}).countByWindow(Seconds(10), Seconds(5))
trans.foreachRDD(rdd=>{
if(!rdd.isEmpty()){
val offsetRanges = A.get("rdd1").get//.asInstanceOf[HasOffsetRanges].offsetRanges
rdd.foreachPartition { iter =>
val o: OffsetRange = offsetRanges(TaskContext.get.partitionId)
println(s"${o.topic} ${o.partition} ${o.fromOffset} ${o.untilOffset}")
}
println(rdd.count())
println(offsetRanges)
// 手動(dòng)提交offset ,前提是禁止自動(dòng)提交
messages.asInstanceOf[CanCommitOffsets].commitAsync(offsetRanges)
}
// A.-("rdd1")
})
// 啟動(dòng)流
ssc.start()
ssc.awaitTermination()
}
def getLastOffsets(kafkaParams : Map[String, Object],topics:Set[String]): Map[TopicPartition, Long] ={
val props = new Properties()
props.putAll(kafkaParams.asJava)
val consumer = new KafkaConsumer[String, String](props)
consumer.subscribe(topics.asJavaCollection)
paranoidPoll(consumer)
val map = consumer.assignment().asScala.map { tp =>
println(tp+"---" +consumer.position(tp))
tp -> (consumer.position(tp))
}.toMap
println(map)
consumer.close()
map
}
def paranoidPoll(c: Consumer[String, String]): Unit = {
val msgs = c.poll(0)
if (!msgs.isEmpty) {
// position should be minimum offset per topicpartition
msgs.asScala.foldLeft(Map[TopicPartition, Long]()) { (acc, m) =>
val tp = new TopicPartition(m.topic, m.partition)
val off = acc.get(tp).map(o => Math.min(o, m.offset)).getOrElse(m.offset)
acc + (tp -> off)
}.foreach { case (tp, off) =>
c.seek(tp, off)
}
}
}
}
到此,關(guān)于“spark streaming窗口聚合操作后怎么管理offset”的學(xué)習(xí)就結(jié)束了,希望能夠解決大家的疑惑。理論與實(shí)踐的搭配能更好的幫助大家學(xué)習(xí),快去試試吧!若想繼續(xù)學(xué)習(xí)更多相關(guān)知識(shí),請(qǐng)繼續(xù)關(guān)注億速云網(wǎng)站,小編會(huì)繼續(xù)努力為大家?guī)?lái)更多實(shí)用的文章!
免責(zé)聲明:本站發(fā)布的內(nèi)容(圖片、視頻和文字)以原創(chuàng)、轉(zhuǎn)載和分享為主,文章觀點(diǎn)不代表本網(wǎng)站立場(chǎng),如果涉及侵權(quán)請(qǐng)聯(lián)系站長(zhǎng)郵箱:is@yisu.com進(jìn)行舉報(bào),并提供相關(guān)證據(jù),一經(jīng)查實(shí),將立刻刪除涉嫌侵權(quán)內(nèi)容。