本文主要是介绍Spark2.x 入门:RDD队列流(DStream),希望对大家解决编程问题提供一定的参考价值,需要的开发者们随着小编来一起学习吧!
在调试Spark Streaming应用程序的时候,我们可以使用streamingContext.queueStream(queueOfRDD)创建基于RDD队列的DStream。
下面是参考Spark官网的QueueStream程序设计的程序,每隔1秒创建一个RDD,Streaming每隔2秒就对数据进行处理。
新建一个TestRDDQueueStream.scala文件,在该文件中输入以下代码:
package org.apache.spark.examples.streaming
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]) {val sparkConf = new SparkConf().setAppName("TestRDDQueue").setMaster("local[2]")val ssc = new StreamingContext(sparkConf, Seconds(20))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()for (i <- 1 to 10){rddQueue += ssc.sparkContext.makeRDD(1 to 100,2)Thread.sleep(1000)}ssc.stop()}
}
打包成功后,执行下面命令运行程序:
spark2-submit --class "org.apache.spark.examples.streaming.QueueStream" /home/songxitang/spark/mycode/streaming/simple-project_2.11-1.0.jar
执行上面命令以后,程序就开始运行,就可以看到类似下面的结果:
-------------------------------------------
Time: 1479522100000 ms
-------------------------------------------
(4,10)
(0,10)
(6,10)
(8,10)
(2,10)
(1,10)
(3,10)
(7,10)
(9,10)
(5,10)
如果无法看到类似上面的屏幕信息,请修改log4j的设置,首先在终端内输入如下命令:
cd /usr/local/spark/conf
vim log4j.properties
打开后,要把其中的rootCategory设置为如下:
log4j.rootCategory=INFO, console
或者,直接把log4j.properties文件删除也可以。
这篇关于Spark2.x 入门:RDD队列流(DStream)的文章就介绍到这儿,希望我们推荐的文章对编程师们有所帮助!