E-MapReduce Spark + MNS是什么?

Spark + MNS



Spark 接入 MNS


下面这个例子演示了 Spark Streaming 如何消费 MNS 中的数据，统计每个 batch 内的单词个数。

2.     val conf = new SparkConf().setAppName("Test MNS Streaming")
3.     val batchInterval = Seconds(10)
4.     val ssc = new StreamingContext(conf, batchInterval)
6.     val queuename = "queuename"
7.     val accessKeyId = "<accessKeyId>"
8.     val accessKeySecret = "<accessKeySecret>"
9.     val endpoint = "http://xxx.yyy.zzzz/abc"
11.     val mnsStream = MnsUtils.createPullingStreamAsRawBytes(ssc, queuename, accessKeyId, accessKeySecret, endpoint,
12.       StorageLevel.MEMORY_ONLY)
13.     mnsStream.foreachRDD( rdd => {
14.       rdd.map(bytes => new String(bytes)).flatMap(line => line.split(" "))
15.         .map(word => (word, 1))
16.         .reduceByKey(_ + _).collect().foreach(e => println(s"word: ${e._1}, cnt: ${e._2}"))
17.     })
19.     ssc.start()
20.     ssc.awaitTermination()

1. MnsUtils.createPullingStreamAsBytes(ssc, queueName, endpoint, storageLevel)
2. MnsUtils.createPullingStreamAsRawBytes(ssc, queueName, endpoint, storageLevel)

• Spark接入 MNS