Simple Log Service:Use Spark Streaming to consume log data

Consume log data in Receiver mode

Consume log data in Direct mode

You must configure the ZooKeeper service when you consume data in Direct mode. This service is used to save the data in the intermediate state. You must set a checkpoint directory in the ZooKeeper service to store the intermediate data. To re-consume data after you restart a task, you must delete the corresponding directory from ZooKeeper and change the name of the consumer group.

• Parameters

  Parameter	Type	Description
  project	String	The name of the project in Simple Log Service.
  logstore	String	The name of the Logstore in Simple Log Service.
  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 Simple Log Service project resides. For more information, see Endpoints.
  accessKeyId	String	The AccessKey ID that is used to access Simple Log Service.
  accessKeySecret	String	The AccessKey secret that is used to access Simple Log Service.
  zkAddress	String	The connection URL of the ZooKeeper service.

• Consumption limits

  Spark Streaming consumes data from each shard in a single batch. You must specify the number of log entries that are consumed in each batch.

  In Simple Log Service, a log group serves as the basic unit for each write request. For example, a write request may contain multiple log entries. These log entries are stored and consumed as a log group. When you use web tracking to write logs, each write request contains only one log entry. In this case, the log group that corresponds to the request contains only one log entry. You can specify parameters to limit the amount of log data in a single batch. The following table lists the two parameters.

  Parameter	Description	Default
  spark.loghub.batchGet.step	The maximum number of log groups that are returned for a single consumption request.	100
  spark.streaming.loghub.maxRatePerShard	The maximum number of log entries that are consumed from each shard in a single batch.	10000

  You can set the maximum number of log entries that are consumed from each shard in each batch by specifying the spark.streaming.loghub.maxRatePerShard parameter. The Spark SDK consumes log data from Simple Log Service by obtaining the number of log groups from the spark.loghub.batchGet.step parameter and accumulating the number of log entries in these log groups. When the accumulated number reaches or exceeds the specified number in the spark.streaming.loghub.maxRatePerShard parameter, the Spark SDK stops consuming log data. The spark.streaming.loghub.maxRatePerShard parameter does not precisely control the number of consumed log entries in each batch. The number of consumed log entries in each batch is based on the spark.loghub.batchGet.step parameter 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 more information, visit GitHub.