Simple Log Service：Spark Streaming消費

Receiver模式

Direct模式

Direct模式需要依賴ZooKeeper環境，用於臨時儲存中間狀態。同時，必須設定Checkpoint目錄。中間狀態資料儲存在ZooKeeper內對應的Checkpoint目錄內。如果任務重啟之後希望重新消費，需要在ZooKeeper內刪除該目錄，並更改消費組名稱。

• 參數說明

  參數	類型	說明
  project	String	Log ServiceProject名稱。
  logstore	String	Log ServiceLogstore名稱。
  consumerGroup	String	消費組名稱，僅用於儲存消費位置。
  endpoint	String	Log ServiceProject所在地區的服務入口。更多資訊，請參見服務存取點。
  accessKeyId	String	訪問Log Service的Access Key ID。
  accessKeySecret	String	訪問Log Service的Access Key Secret。
  zkAddress	String	ZooKeeper的串連地址。

• 限流配置

  Spark Streaming流式消費是將資料分成微小的批次進行處理，因此Log Service開始消費時，需預知每個批次的邊界，即每個批次需要擷取的資料條數。

  Log Service底層的儲存模型以LogGroup為單位，正常情況下每個LogGroup對應一次寫入請求，例如一次寫入請求可能包含數千條日誌，這些日誌作為一個LogGroup進行儲存和消費。而通過WebTracking方式寫入日誌時，每次寫入請求僅包含一條日誌，即一個LogGroup只有一條日誌。為了能夠應對不同寫入情境的消費需求，SDK提供如下兩個參數進行限流配置。

  參數	說明	預設值
  spark.loghub.batchGet.step	限制單次消費請求擷取的最大LogGroup個數。	100
  spark.streaming.loghub.maxRatePerShard	限制單批次內每個Shard消費的日誌條數。	10000

  通過spark.streaming.loghub.maxRatePerShard可指定每個批次每個Shard期望消費的最大日誌條數。Spark SDK的實現原理是每次從服務端擷取spark.loghub.batchGet.step中的LogGroup個數並累計其中的日誌條數，直到達到或超過spark.streaming.loghub.maxRatePerShard。因此spark.streaming.loghub.maxRatePerShard並非一個精確控制單批次消費日誌條數的參數，實際上每個批次消費的日誌條數與spark.loghub.batchGet.step以及每個LogGroup中包含的日誌條數相關。

• 樣本

  import com.aliyun.openservices.loghub.client.config.LogHubCursorPosition
  import org.apache.spark.SparkConf
  import org.apache.spark.streaming.{Milliseconds, StreamingContext}
  import org.apache.spark.streaming.aliyun.logservice.{CanCommitOffsets, LoghubUtils}
  
  object TestDirectLoghub {
    def main(args: Array[String]): Unit = {
      if (args.length < 7) {
        System.err.println(
          """Usage: TestDirectLoghub <project> <logstore> <loghub group name> <endpoint>
            |         <access key id> <access key secret> <batch interval seconds> <zookeeper host:port=localhost:2181>
          """.stripMargin)
        System.exit(1)
      }
  
      val project = args(0)
      val logstore = args(1)
      val consumerGroup = args(2)
      val endpoint = args(3)
      val accessKeyId = args(4)
      val accessKeySecret = args(5)
      val batchInterval = Milliseconds(args(6).toInt * 1000)
      val zkAddress = if (args.length >= 8) args(7) else "localhost:2181"
  
      def functionToCreateContext(): StreamingContext = {
        val conf = new SparkConf().setAppName("Test Direct Loghub")
        val ssc = new StreamingContext(conf, batchInterval)
        val zkParas = Map("zookeeper.connect" -> zkAddress,
          "enable.auto.commit" -> "false")
        val loghubStream = LoghubUtils.createDirectStream(
          ssc,
          project,
          logstore,
          consumerGroup,
          accessKeyId,
          accessKeySecret,
          endpoint,
          zkParas,
          LogHubCursorPosition.END_CURSOR)
  
        loghubStream.checkpoint(batchInterval).foreachRDD(rdd => {
          println(s"count by key: ${rdd.map(s => {
            s.sorted
            (s.length, s)
          }).countByKey().size}")
          loghubStream.asInstanceOf[CanCommitOffsets].commitAsync()
        })
        ssc.checkpoint("hdfs:///tmp/spark/streaming") // set checkpoint directory
        ssc
      }
  
      val ssc = StreamingContext.getOrCreate("hdfs:///tmp/spark/streaming", functionToCreateContext _)
      ssc.start()
      ssc.awaitTermination()
    }
  }

更多資訊，請參見GitHub。