最近在做實時流處理的一個專案,遇到N多問題,經過不斷的除錯,終於有點進展,記錄一下,防止後人遇到同樣的問題.

1,sparkstreaming消費kafka有兩種方法,這裡我就不介紹了,網上關於這方面的資料很多,我就簡單說一下兩者的區別吧,

(1)基於receiver的方式，是使用Kafka的高階API來在ZooKeeper中儲存消費過的offset的。這是消費Kafka資料的傳統方式。這種方式配合著WAL機制可以保證資料零丟失的高可靠性，但是卻無法保證資料被處理一次且僅一次，可能會處理兩次。因為Spark和ZooKeeper之間可能是不同步的.

(2)基於direct的方式，使用kafka的簡單api，Spark Streaming自己就負責追蹤消費的offset，並儲存在checkpoint中。Spark自己一定是同步的，因此可以保證資料是消費一次且僅消費一次,在實際生產環境中大都用Direct方式.

特別說明一下:kafka0.9之前,offest都是儲存在zk中,有zk來維護的,0.9之後kafka的offest儲存在kafka的 __consumer_offsets 這個topic中了.

2,手動維護kafka的offest.

(1),為了實現exactly-once的語義,我採用自己儲存offest的方法,offest可以儲存在zk,kafka,mysql,hbase,redis中自己根據情況而定,我選擇把offest儲存到redis中.建立Dstream之前,先判斷是否消費過,如果沒有消費就從頭開始,如果已經消費過了,就從上次儲存的offest處開始消費,廢話不多說,直接上程式碼.(因程式碼有點多,就只貼了重要的部分)

(2),spark版本2.2.0,scala版本2.11.8,kafka版本0.10.1,hbase版本1.1.2.

程式碼如下:

package test

import java.util
import kafka.{PropertiesScalaUtils, RedisKeysListUtils}
import kafka.SparkStreamingKafka.{dbIndex, kafkaStreams}
import org.apache.kafka.common.TopicPartition
import org.apache.kafka.common.serialization.StringDeserializer
import org.apache.log4j.{Level, Logger}
import org.apache.spark.SparkConf
import org.apache.spark.streaming.{Seconds, StreamingContext}
import org.apache.spark.streaming.kafka010.{ConsumerStrategies, KafkaUtils, LocationStrategies}
import redis.RedisPool

object sparkstreaming {
  def main(args: Array[String]): Unit = {
    Logger.getLogger("org.apache.spark").setLevel(Level.INFO)
    Logger.getLogger("org.eclipse.jetty.server").setLevel(Level.INFO)
    Logger.getLogger("org.apache.kafka.clients.consumer").setLevel(Level.INFO)
    val conf = new SparkConf().setAppName("sparkstreaming")
    conf.set("spark.streaming.kafka.maxRatePerPartition", "2000")
    conf.set("spark.serializer", "org.apache.spark.serializer.KryoSerializer")
    conf.set("spark.streaming.concurrentJobs", "10")
    conf.set("spark.streaming.kafka.maxRetries", "50")
    val scc = new StreamingContext(conf, Seconds(5))
    val topic = PropertiesScalaUtils.loadProperties("topic")
    val topicSet: Set[String] = Set(topic)
    val kafkaParams = Map[String, Object](
      "auto.offset.reset" -> "latest",
      "value.deserializer" -> classOf[StringDeserializer]
      , "key.deserializer" -> classOf[StringDeserializer]
      , "bootstrap.servers" -> PropertiesScalaUtils.loadProperties("broker")
      , "group.id" -> PropertiesScalaUtils.loadProperties("groupId")
      , "enable.auto.commit" -> (false: java.lang.Boolean)
    )
    val maxTotal = 200
    val maxIdle = 100
    val minIdle = 10
    val testOnBorrow = false
    val testOnReturn = false
    val maxWaitMillis = 500
    RedisPool.makePool(PropertiesScalaUtils.loadProperties("redisHost"), PropertiesScalaUtils.loadProperties("redisPort").toInt, maxTotal, maxIdle, minIdle, testOnBorrow, testOnReturn, maxWaitMillis)
    val jedis = RedisPool.getPool.getResource
    jedis.select(dbIndex)
    val keys: util.Set[String] = jedis.keys(topic + "*")
    if (keys.size() == 0) {
      kafkaStreams = KafkaUtils.createDirectStream[String, String](
        scc,
        LocationStrategies.PreferConsistent,
        ConsumerStrategies.Subscribe[String, String](topicSet, kafkaParams))
    } else {
      val fromOffsets: Map[TopicPartition, Long] = RedisKeysListUtils.getKeysList(PropertiesScalaUtils.loadProperties("redisHost"), PropertiesScalaUtils.loadProperties("redisPort").toInt, topic)
      kafkaStreams = KafkaUtils.createDirectStream[String, String](
        scc,
        LocationStrategies.PreferConsistent,
        ConsumerStrategies.Subscribe[String, String](topicSet, kafkaParams, fromOffsets))
    }
    RedisPool.getPool.returnResource(jedis)
    kafkaStreams.foreachRDD(rdd=>{
      val offsetRanges = rdd.asInstanceOf[HasOffsetRanges].offsetRanges
      rdd.foreachPartition(partiton=>{
        val conf = HBaseConfiguration.create()
        conf.set("hbase.zookeeper.quorum", PropertiesScalaUtils.loadProperties("zk_hbase")) //zk的地址;
        conf.set("hbase.zookeeper.property.clientPort", PropertiesScalaUtils.loadProperties("zk_port"))
        conf.set("hbase.master", PropertiesScalaUtils.loadProperties("hbase_master"))
        conf.set("hbase.defaults.for.version.skip", "true")
        conf.set("hhbase.rootdir", PropertiesScalaUtils.loadProperties("hbase_rootdir"))
        conf.set("zookeeper.znode.parent", PropertiesScalaUtils.loadProperties("zookeeper_znode_parent"))
        myTable = new HTable(conf, TableName.valueOf(PropertiesScalaUtils.loadProperties("hbase_table")))
        myTable.setAutoFlush(false, false) //關閉自動提交
        myTable.setWriteBufferSize(3 * 1024 * 1024)
        partiton.foreach(pair=>{
          //自己的處理邏輯;
        })
        myTable.flushCommits()
        myTable.close()
        offsetRanges.foreach { offsetRange =>
          println("partition : " + offsetRange.partition + " fromOffset:  " + offsetRange.fromOffset + " untilOffset: " + offsetRange.untilOffset)
          val topic_partition_key_new = offsetRange.topic + "_" + offsetRange.partition
          jedis_jason.set(topic_partition_key_new, offsetRange.untilOffset + "")
        })
      })
      scc.start()
      scc.awaitTermination()
    }
  }

 

程式碼已經測試過了,沒有問題,如果有寫的不對的地方,歡迎大家指正,如果有什麼疑問,可以加QQ群:340297350,謝謝