E-MapReduce:Integrate Spark with

Prerequisites

Before you begin, ensure that you have:

• An SMQ queue with at least one message to consume

• An EMR cluster with Spark Streaming enabled

• The EMR SDK (version 1.3.2 or later for MetaService access)

• (If using AccessKey authentication) Environment variables ALIBABA_CLOUD_ACCESS_KEY_ID and ALIBABA_CLOUD_ACCESS_KEY_SECRET configured on your cluster nodes. See Configure environment variables.

Authentication options

EMR SDK provides two ways to authenticate Spark Streaming against SMQ:

Use MetaService when running on an EMR cluster — it removes the need to manage or rotate AccessKey credentials in your code.

Consume SMQ messages with MetaService

MetaService handles authentication automatically. Pass the queue name, endpoint, and storage level — no AccessKey pair required.

val conf = new SparkConf().setAppName("Test MNS Streaming")
val batchInterval = Seconds(10)
val ssc = new StreamingContext(conf, batchInterval)

val queueName = "<your-queue-name>"
// Use the SMQ endpoint for your region
val endpoint = "https://oss-cn-hangzhou.aliyuncs.com"

// MetaService authenticates automatically — no AccessKey pair needed
val mnsStream = MnsUtils.createPullingStreamAsBytes(ssc, queueName, endpoint, StorageLevel.MEMORY_ONLY)

mnsStream.foreachRDD { rdd =>
  rdd.map(bytes => new String(bytes))
    .flatMap(_.split(" "))
    .map(word => (word, 1))
    .reduceByKey(_ + _)
    .collect()
    .foreach(e => println(s"word: ${e._1}, cnt: ${e._2}"))
}

ssc.start()
ssc.awaitTermination()

Replace the following placeholders with actual values:

Consume SMQ messages with an AccessKey pair

If MetaService is unavailable, pass your AccessKey credentials explicitly. Store them as environment variables rather than hardcoding them in your code.

val conf = new SparkConf().setAppName("Test MNS Streaming")
val batchInterval = Seconds(10)
val ssc = new StreamingContext(conf, batchInterval)

val queuename = "<your-queue-name>"
val accessKeyId = System.getenv("ALIBABA_CLOUD_ACCESS_KEY_ID")
val accessKeySecret = System.getenv("ALIBABA_CLOUD_ACCESS_KEY_SECRET")
// In this example, the endpoint of the China (Hangzhou) region is used. Specify your actual endpoint.
val endpoint = "https://oss-cn-hangzhou.aliyuncs.com"

val mnsStream = MnsUtils.createPullingStreamAsRawBytes(ssc, queuename, accessKeyId, accessKeySecret, endpoint,
  StorageLevel.MEMORY_ONLY)

mnsStream.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()

What's next

For the complete sample code, see SparkMNSDemo on GitHub.