E-MapReduce Spark + ONS是什么?

Spark + ONS



Spark 接入 ONS


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

1.     val Array(cId, topic, subExpression, parallelism, interval) = args
3.     val accessKeyId = "<accessKeyId>"
4.     val accessKeySecret = "<accessKeySecret>"
6.     val numStreams = parallelism.toInt
7.     val batchInterval = Milliseconds(interval.toInt)
9.     val conf = new SparkConf().setAppName("Test ONS Streaming")
10.     val ssc = new StreamingContext(conf, batchInterval)
11.     def func: Message => Array[Byte] = msg => msg.getBody
12.     val onsStreams = (0 until numStreams).map { i =>
13.       println(s"starting stream $i")
14.       OnsUtils.createStream(
15.         ssc,
16.         cId,
17.         topic,
18.         subExpression,
19.         accessKeyId,
20.         accessKeySecret,
21.         StorageLevel.MEMORY_AND_DISK_2,
22.         func)
23.     }
25.     val unionStreams = ssc.union(onsStreams)
26.     unionStreams.foreachRDD(rdd => {
27.       rdd.map(bytes => new String(bytes)).flatMap(line => line.split(" "))
28.         .map(word => (word, 1))
29.         .reduceByKey(_ + _).collect().foreach(e => println(s"word: ${e._1}, cnt: ${e._2}"))
30.     })
32.     ssc.start()
33.     ssc.awaitTermination()

• Spark接入ONS