Simple Log Service:Spark Streaming consumption

Limitations

The Spark Streaming consumer supports only Spark 2.x versions.

Receiver mode

Direct mode

Direct mode requires a ZooKeeper cluster to store intermediate state. You must specify a checkpoint directory path in ZooKeeper for this data. To re-consume data after a job restart, delete this directory in ZooKeeper and change the consumer group name.

• Parameters

  Parameter	Type	Description
  Project	String	The name of the Simple Log Service Project.
  LogStore	String	The name of the LogStore.
  consumerGroup	String	The name of the consumer group, used only to save the consumption checkpoint.
  endpoint	String	The endpoint for the region where the Simple Log Service Project is located. For more information, see Endpoints.
  accessKeyId	String	The AccessKey ID used to access Simple Log Service.
  accessKeySecret	String	The AccessKey Secret used to access Simple Log Service.
  zkAddress	String	The connection address of your ZooKeeper cluster.

• Rate limiting configuration

  Spark Streaming processes data in micro-batches. For each batch, the consumer must determine how many log entries to fetch.

  The underlying storage model of Simple Log Service uses the LogGroup as a unit. Normally, each LogGroup corresponds to a single write request and can contain thousands of logs. However, in scenarios like web tracking, each write request might contain only a single log. To handle these different scenarios, the SDK provides two rate-limiting parameters:

  Parameter	Description	Default
  spark.loghub.batchGet.step	The maximum number of LogGroups to fetch in a single request.	100
  spark.streaming.loghub.maxRatePerShard	The maximum number of log entries to consume from a single shard per batch.	10000

  The parameter spark.streaming.loghub.maxRatePerShard sets the target for the maximum number of log entries to fetch from a shard in each batch. The SDK fetches LogGroups in increments defined by spark.loghub.batchGet.step until the total log count meets or exceeds the target set by spark.streaming.loghub.maxRatePerShard. Because of this, spark.streaming.loghub.maxRatePerShard acts as a soft limit. The actual number of logs fetched per batch will vary depending on spark.loghub.batchGet.step and the number of logs within each LogGroup.

• 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, see the project on GitHub.