kafka結合streaming的兩種方式

SparkStreaming的Receiver方式和直連方式的區別？Receiver接收固定時間間隔的數據(放在內存中的)，使用kafka高級API，自動維護偏移量， 達到固定時間才能進行處理，效率低下（），並且容易丟失數據 Direct直連方式，相當於直連連接到Kafka的分區上，使用Kafka底層API，效率高，需要自己維護偏移量 Receiver方式和直連的方式： 如果說這兩種方式設置的時間間隔是5秒的話，Receiver方式拉去的數據是等待5秒，然后將這5秒的數據 全部一次性拿過來進行操作，這樣等待5秒的時間，如果數據量特變大的情況下，杯子容量不夠，會造成數據丟失。 Direct直連方式，這種方式是直接連接到kafka對應的分區上，如果kafka中有數據的話，就直接從kafka中拉取數據， 不用等待5秒的時間才拉取數據，等到了5秒的時間才去執行一次數據。

 

1.Receiver方式：

 

 

 

這種方式的缺點在於：如果5秒鍾的時間，receiver去kafka中拉數據，如果數據量小的話， 是可以接受的但是當數據量大的時候，5秒的時間可能拉取了大量的數據，receiver可能已經不能容納下這么大量的數據，就會導致數據的丟失問題，這里數據的偏移量，既update offset 是交給zookeeper管理的，這種方式在生產過程中，已經很少有使用了，甚至在kafka0-10以后就直接將receiver移除了。這里的receiver就相當於一個杯子，這個回去kafka的流中接收數據，杯子的容量是一定的，當kafka流中的量過大的時候，就會導致數據的丟失。接收5秒鍾才進行處理 ，這樣的效率比較慢。 如果在這5秒的時間拉取過來的數據量過大，因為內存中的大小是有限的，內存中的數據會溢出。 receiver方式：它使用的是高級API實現Offset自動管理，不需要我們去管理，那么這樣的話，它的靈活性特別差，不好，而且處理數據的時候，如果某一時刻的數據量過大，那么會造成數據丟失的情況，為了防止數據的丟失 ，在streaming1.2之后就加入了WAL機制，就是寫前日志，將從kafka中讀取的數據先保存到hdfs或者亞馬遜的S3上，然后再去消費數據，WAL是為了防止數據的丟失，可以對數據進行恢復。 2.直連的方式(Direct) 直連方式：他使用的是底層API實現Offset，我們開發人員管理，這樣的話，它的靈活性很好，並且可以保證數據的安全性，而且不用擔心數據量過大，因為他有浴池里機制，進行提前處理，然后批次提交任務。

 

 

舍棄了WAL的方式，使用更加底層的api，偏移量需要自己去查詢和管理，先通過自己維護的偏移量去讀取數據，這種方式數據不會經過磁盤，這里會有一個預處理機制，相當於直接用水管插到kafka上，當拉去到數據之后就直接進行預處理機制產生結果集，然后到達到設置的時間后，把所有預處理機制產生的結果匯集起來，這樣會減少數據的大小，因為每次來的數據都會進行一個預處理，這里的預處理也就是執行計算邏輯，當達到預定時間，就直接將預處理的數據進行一個匯總。 代碼如下：

def main(args: Array[String]): Unit = {
    //指定組名
    val group = "g1"
    //創建SparkConf
    val conf = new SparkConf().setAppName("KafkaDirectWordCount").setMaster("local[2]")
    //創建SparkStreaming，並設置間隔時間
    val ssc = new StreamingContext(conf, Duration(5000))
    //指定消費的 topic 名字
    val topic = "test4"
    //指定kafka的broker地址(sparkStream的Task直連到kafka的分區上，用更加底層的API消費，效率更高)
    val brokerList = "mini4:9092,mini5:9092,mini6:9092"
    val broadcastRef: Broadcast[Array[(Long, Long, String)]] = IPUtils.broadcastIpRules(ssc, "hdfs://mini4:9000/date/ip.txt")
    //指定zk的地址，后期更新消費的偏移量時使用(以后可以使用Redis、MySQL來記錄偏移量)
    val zkQuorum = "mini4:2181,mini5:2181,mini5:2181"
    //創建 stream 時使用的 topic 名字集合，SparkStreaming可同時消費多個topic
    val topics: Set[String] = Set(topic)

    //創建一個 ZKGroupTopicDirs 對象,其實是指定往zk中寫入數據的目錄，用於保存偏移量
    val topicDirs = new ZKGroupTopicDirs(group, topic)
    //獲取 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]()

    //直連方式只有在KafkaDStream的RDD中才能獲取偏移量，那么就不能到調用DStream的Transformation
    //所以只能子在kafkaStream調用foreachRDD，獲取RDD的偏移量，然后就是對RDD進行操作了
    //依次迭代KafkaDStream中的KafkaRDD
    kafkaStream.foreachRDD { kafkaRDD =>
      if (!kafkaRDD.isEmpty()) {
        //只有KafkaRDD可以強轉成HasOffsetRanges，並獲取到偏移量
        offsetRanges = kafkaRDD.asInstanceOf[HasOffsetRanges].offsetRanges
        //拿到一個rdd
        val lines: RDD[String] = kafkaRDD.map(_._2)
        //這里的lines就是RDD了。
        //對RDD進行操作，觸發Action
        val lineRDD: RDD[Array[String]] = lines.map(_.split(" "))
        //計算成交總金額
        CalculateUtil.calculateIncome(lineRDD)
        //計算商品分類金額
        CalculateUtil.calculateItem(lineRDD)
        //計算區域總和
        CalculateUtil.calculateZone(lineRDD,broadcastRef)
        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()
  }