Consume data from ApsaraMQ for RocketMQ and calculate word count with Spark Streaming - E-MapReduce

Use Spark Streaming on E-MapReduce (EMR) to consume messages from ApsaraMQ for RocketMQ. The following example subscribes to a topic, receives messages across parallel streams, and runs a word count on each batch.

Prerequisites

Before you begin, ensure that you have:

An EMR cluster with Spark Streaming enabled
A topic and consumer group configured in ApsaraMQ for RocketMQ
An AccessKey ID and AccessKey secret with permission to read from the topic
Environment variables set for your credentials (see Configure environment variables)

How Spark Streaming connects to ApsaraMQ for RocketMQ

The job creates multiple Spark Streaming receivers — one per parallel stream — and merges them into a single DStream using ssc.union(). Each batch of incoming messages is decoded from bytes to strings, split into words, and aggregated with reduceByKey. The word counts are printed to the console after each batch.

OnsUtils.createStream() handles the connection to ApsaraMQ for RocketMQ. The following table describes its parameters:

Parameter	Description	Example
`ssc`	The `StreamingContext` instance	—
`cId`	Consumer group ID	`GID_your-group`
`topic`	Topic name to subscribe to	`your-topic`
`subExpression`	Subscription expression for message filtering	`*` (all messages)
`accessKeyId`	AccessKey ID, read from environment variable `ALIBABA_CLOUD_ACCESS_KEY_ID`	—
`accessKeySecret`	AccessKey secret, read from environment variable `ALIBABA_CLOUD_ACCESS_KEY_SECRET`	—
`storageLevel`	RDD storage level for received data	`StorageLevel.MEMORY_AND_DISK_2`
`func`	Function to extract the message body as bytes	`msg => msg.getBody`

The job accepts five arguments at runtime:

Argument	Description
`cId`	Consumer group ID
`topic`	Topic name
`subExpression`	Subscription expression
`parallelism`	Number of parallel receiver streams
`interval`	Batch interval in milliseconds

Sample code

val Array(cId, topic, subExpression, parallelism, interval) = args
val accessKeyId = System.getenv("ALIBABA_CLOUD_ACCESS_KEY_ID")
val accessKeySecret = System.getenv("ALIBABA_CLOUD_ACCESS_KEY_SECRET")
val numStreams = parallelism.toInt
val batchInterval = Milliseconds(interval.toInt)
val conf = new SparkConf().setAppName("Test ONS Streaming")
val ssc = new StreamingContext(conf, batchInterval)
def func: Message => Array[Byte] = msg => msg.getBody
val onsStreams = (0 until numStreams).map { i =>
  println(s"starting stream $i")
  OnsUtils.createStream(
    ssc,
    cId,
    topic,
    subExpression,
    accessKeyId,
    accessKeySecret,
    StorageLevel.MEMORY_AND_DISK_2,
    func)
}
val unionStreams = ssc.union(onsStreams)
unionStreams.foreachRDD(rdd => {
  rdd.map(bytes => new String(bytes)).flatMap(line => line.split(" "))
    .map(word => (word, 1))
    .reduceByKey(_ + _).collect().foreach(e => println(s"word: ${e._1}, cnt: ${e._2}"))
})
ssc.start()
ssc.awaitTermination()

Note

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

What's next

For the complete sample code, see the SparkRocketMQDemo example on GitHub.