Consume data in Simple Log Service

Prerequisites

Before you begin, ensure that you have:

• An EMR cluster with Spark Streaming enabled

• A Simple Log Service project and Logstore with log data

• An AccessKey ID and AccessKey secret with read access to the Logstore, or MetaService enabled on the cluster (EMR SDK V1.3.2 and later)

• The internal endpoint of Simple Log Service — all EMR cluster nodes except the master node cannot connect to the internet

Choose an access method

The Direct API-based DStream method is experimental. Review the constraints in that section before using it in production.

Method 1: Receiver-based DStream

Each receiver reads from one or more shards in the Logstore. The number of receivers controls parallelism.

val logServiceProject = args(0)       // Simple Log Service project name
val logStoreName = args(1)            // Logstore name
val loghubConsumerGroupName = args(2) // Consumer group name
val loghubEndpoint = args(3)          // Internal endpoint of Simple Log Service
val accessKeyId = System.getenv("ALIBABA_CLOUD_ACCESS_KEY_ID")
val accessKeySecret = System.getenv("ALIBABA_CLOUD_ACCESS_KEY_SECRET")
val numReceivers = args(4).toInt      // Number of receivers
val batchInterval = Milliseconds(args(5).toInt * 1000)

val conf = new SparkConf().setAppName("Test Loghub Streaming")
val ssc = new StreamingContext(conf, batchInterval)
val loghubStream = LoghubUtils.createStream(
  ssc,
  logServiceProject,
  logStoreName,
  loghubConsumerGroupName,
  loghubEndpoint,
  numReceivers,
  accessKeyId,
  accessKeySecret,
  StorageLevel.MEMORY_AND_DISK)

loghubStream.foreachRDD(rdd => println(rdd.count()))
ssc.start()
ssc.awaitTermination()

Parameters

Set ALIBABA_CLOUD_ACCESS_KEY_ID and ALIBABA_CLOUD_ACCESS_KEY_SECRET as environment variables before running the job. For instructions, see Configure environment variables.

Method 2: Direct API-based DStream (experimental)

Available in EMR SDK V1.4.0 and later. This method reads directly from the Logstore. Data in LogHub is not repeatedly stored as write-ahead logging (WAL) files. It allows you to write data at least once without the need to enable the WAL feature of Spark Streaming.

val logServiceProject = args(0)
val logStoreName = args(1)
val loghubConsumerGroupName = args(2)
val loghubEndpoint = args(3)
val accessKeyId = args(4)
val accessKeySecret = args(5)
val batchInterval = Milliseconds(args(6).toInt * 1000)
val zkConnect = args(7)
val checkpointPath = args(8)

def functionToCreateContext(): StreamingContext = {
  val conf = new SparkConf().setAppName("Test Direct Loghub Streaming")
  val ssc = new StreamingContext(conf, batchInterval)
  val zkParas = Map("zookeeper.connect" -> zkConnect, "enable.auto.commit" -> "false")
  val loghubStream = LoghubUtils.createDirectStream(
    ssc,
    logServiceProject,
    logStoreName,
    loghubConsumerGroupName,
    accessKeyId,
    accessKeySecret,
    loghubEndpoint,
    zkParas,
    LogHubCursorPosition.END_CURSOR)
  ssc.checkpoint(checkpointPath)
  val stream = loghubStream.checkpoint(batchInterval)
  stream.foreachRDD(rdd => {
    println(rdd.count())
    loghubStream.asInstanceOf[CanCommitOffsets].commitAsync()
  })
  ssc
}

val ssc = StreamingContext.getOrCreate(checkpointPath, functionToCreateContext _)
ssc.start()
ssc.awaitTermination()

Parameters

Access without an AccessKey pair: MetaService

In EMR SDK V1.3.2 and later, LoghubUtils.createStream supports MetaService, which reads credentials from the cluster automatically. No AccessKey pair is needed in the application code.

// Minimal — uses END_CURSOR by default
LoghubUtils.createStream(ssc, logServiceProject, logStoreName, loghubConsumerGroupName, storageLevel)

// With explicit receiver count
LoghubUtils.createStream(ssc, logServiceProject, logStoreName, loghubConsumerGroupName, numReceivers, storageLevel)

// With cursor position control
LoghubUtils.createStream(ssc, logServiceProject, logStoreName, loghubConsumerGroupName, storageLevel, cursorPosition, mLoghubCursorStartTime, forceSpecial)

// With receiver count and cursor position control
LoghubUtils.createStream(ssc, logServiceProject, logStoreName, loghubConsumerGroupName, numReceivers, storageLevel, cursorPosition, mLoghubCursorStartTime, forceSpecial)

For details on the LoghubUtils class, see the EMR SDK reference.

Cursor positions

The cursor position controls where in the Logstore consumption starts. All modes respect existing checkpoints — if a checkpoint exists, consumption resumes from the checkpoint regardless of the cursor position setting.

To force consumption from a specific timestamp regardless of any existing checkpoint, set the following parameters in createStream:

• cursorPosition: LogHubCursorPosition.SPECIAL_TIMER_CURSOR

• forceSpecial: true

What's next

• Download the complete sample code on GitHub for a full working example.