Spark大数据-输入源之RDD队列流

RDD队列流

每隔一秒创建一个RDD扔到队列中，spark streaming每隔两秒从队列流中处理一次，对队列RDD中的每个数做余数词频统计。

// 每隔1秒创建一个RDD，Streaming每隔2秒就对数据进行处理
import org.apache.spark.SparkConf
import org.apache.spark.rdd.RDD
import org.apache.spark.streaming.StreamingContext._
import org.apache.spark.streaming.{Seconds, StreamingContext}

object QueueStream {
  def main(args: Array[String]) {
    StreamingExamples.setStreamingLogLevels()
//       创建ssc
    val sparkConf = new SparkConf().setAppName("TestRDDQueue").setMaster("local[2]")
    val ssc = new StreamingContext(sparkConf, Seconds(2))
//      创建RDD队列
    val rddQueue =new scala.collection.mutable.SynchronizedQueue[RDD[Int]]()
//      创建队列输入流
    val queueStream = ssc.queueStream(rddQueue)     
    val mappedStream = queueStream.map(r => (r % 10, 1))
    val reducedStream = mappedStream.reduceByKey(_ + _)
    reducedStream.print()
    ssc.start()
//       往队列流中添加RDD
    for (i <- 1 to 10){
// 1~100,生成一个RDD，2个分区
        rddQueue += ssc.sparkContext.makeRDD(1 to 100,2)
        Thread.sleep(1000)
    }
    ssc.stop()
  }
}
QueueStream.main(Array())

打包编译运行过程类似于套接字流。