438      阿裏雲

Spark + Log Service__Spark_開發人員指南_E-MapReduce-阿裏雲

Spark + Log Service

Spark 接入 Log Service

下麵這個例子演示了 Spark Streaming 如何消費 Log Service 中的日誌數據，統計日誌條數。

    if (args.length < 6) {
      System.err.println(
        """Usage: TestLoghub <sls project> <sls logstore> <loghub group name>  
          |         <sls endpoint> <receiver number> <batch interval seconds>
        """.stripMargin)
      System.exit(1)
    }
    val logserviceProject = args(0)    // Log Service 中 project 名
    val logStoreName = args(1)     // Log Service 中 logstore 名
    val loghubGroupName = args(2)  // loghubGroupName 相同的作業將共同消費 logstore 的數據
    val loghubEndpoint = args(3)  // 阿裏雲日誌服務數據類 API Endpoint
    val accessKeyId = "<accessKeyId>"     // 訪問日誌服務的 AccessKeyId
    val accessKeySecret = "<accessKeySecret>" // 訪問日誌服務的 AccessKeySecret
    val numReceivers = args(4).toInt  // 啟動多少個 Receiver 來讀取 logstore 中的數據
    val batchInterval = Milliseconds(args(5).toInt * 1000) // Spark Streaming 中每次處理批次時間間隔
    val conf = new SparkConf().setAppName("Test Loghub Streaming")
    val ssc = new StreamingContext(conf, batchInterval)
    val loghubStream = LoghubUtils.createStream(
      ssc,
      loghubProject,
      logStream,
      loghubGroupName,
      endpoint,
      numReceivers,
      accessKeyId,
      accessKeySecret,
      StorageLevel.MEMORY_AND_DISK)
    loghubStream.foreachRDD(rdd => println(rdd.count()))
    ssc.start()
    ssc.awaitTermination()

說明

• E-MapReduce SDK支持LogService的三種消費模式，即“BEGIN_CURSOR”，“END_CURSOR”和“SPECIAL_TIMER_CURSOR”，默認是“END_CURSOR”。
  • BEGIN_CURSOR：從日誌頭開始消費，如果有checkpoint記錄，則從checkpoint處開始消費。
  • END_CURSOR：從日誌尾開始消費，如果有checkpoint記錄，則從checkpoint處開始消費。
  • SPECIAL_TIMER_CURSOR：從指定時間點開始消費，如果有checkpoint記錄，則從checkpoint處開始消費。單位為秒。
  • 以上三種消費模式都收到checkpoint記錄的影響，如果存在checkpoint記錄，則從checkpoint處開始消費，不管指定的是什麼消費模式。E-MapReduce SDK基於“SPECIAL_TIMER_CURSOR”模式支持用戶強製在指定時間點開始消費：在LoghubUtils#createStream接口中，以下參數需要組合使用：
    • cursorPosition：LogHubCursorPosition.SPECIAL_TIMER_CURSOR
    • forceSpecial：true
• E-MapReduce 的機器（除了 Master 節點）無法連接公網。配置 Log Service endpoint 時，請注意使用 Log Service 提供的內網 endpoint，否則無法請求到 Log Service。
• 了解更多關於 Log Service，請查看相關文檔。

附錄

完整示例代碼請看: