Simple Log Service:Use Spark Streaming to consume log data

Limits

Spark Streaming consumption supports only Spark 2.x.

Consume log data in Receiver mode

Consume log data in Direct mode

Direct mode requires a ZooKeeper service to store intermediate state. Set a checkpoint directory in ZooKeeper to persist the intermediate data. To re-consume data after restarting a task, delete the corresponding ZooKeeper directory and change the consumer group name.

• Parameters

  Parameter	Type	Description
  project	String	The name of the project in SLS.
  logstore	String	The name of the logstore in SLS.
  consumerGroup	String	The name of the consumer group. This name is used only to save consumption checkpoints.
  endpoint	String	The endpoint of the region where the SLS project resides. For more information, see Endpoints.
  accessKeyId	String	The AccessKey ID that is used to access SLS.
  accessKeySecret	String	The AccessKey secret that is used to access SLS.
  zkAddress	String	The connection URL of the ZooKeeper service.

• Rate limiting

  Spark Streaming processes data in micro-batches. Specify the number of log entries consumed per batch per shard.

  In SLS, each write request is stored as a log group. A typical write request contains multiple log entries in one log group. When web tracking is used, each write request contains only one log entry per log group. The following parameters control the amount of data consumed in a single batch.

  Parameter	Description	Default
  spark.loghub.batchGet.step	The maximum number of log groups fetched per consumption request.	100
  spark.streaming.loghub.maxRatePerShard	The maximum number of log entries consumed per shard per batch.	10000

  The spark.streaming.loghub.maxRatePerShard parameter sets the target maximum log entries per shard per batch. The SDK fetches log groups in increments of spark.loghub.batchGet.step and accumulates the log entry count. When the count reaches or exceeds spark.streaming.loghub.maxRatePerShard, the SDK stops fetching. Because consumption granularity is at the log group level, the spark.streaming.loghub.maxRatePerShard limit is approximate. The actual count per batch depends on spark.loghub.batchGet.step and the number of log entries in each log group.

• Example

  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()
    }
  }

For the complete source code, see GitHub.