Spark2.x 入门:RDD队列流(DStream)
2018-01-26 14:53
495 查看
在调试Spark Streaming应用程序的时候,我们可以使用streamingContext.queueStream(queueOfRDD)创建基于RDD队列的DStream。
下面是参考Spark官网的QueueStream程序设计的程序,每隔1秒创建一个RDD,Streaming每隔2秒就对数据进行处理。
新建一个TestRDDQueueStream.scala文件,在该文件中输入以下代码:
打包成功后,执行下面命令运行程序:
执行上面命令以后,程序就开始运行,就可以看到类似下面的结果:
如果无法看到类似上面的屏幕信息,请修改log4j的设置,首先在终端内输入如下命令:
打开后,要把其中的rootCategory设置为如下:
或者,直接把log4j.properties文件删除也可以。
下面是参考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文件删除也可以。
相关文章推荐
- Spark RDD入门详解
- Spark RDD/Core 编程 API入门系列 之rdd实战(rdd基本操作实战及transformation和action流程图)(源码)(三)
- SPARK入门--RDD编程
- Spark2.x 快速入门教程 4
- Spark RDD/Core 编程 API入门系列之简单移动互联网数据(五)
- Spark入门(四):RDD基本操作
- spark2.x由浅入深深到底系列六之RDD java api详解一
- spark2.x由浅入深深到底系列六之RDD java api用JdbcRDD读取关系型数据库
- Spark2.x 快速入门教程 5
- Spark2.x学习笔记:11、RDD依赖关系与stage划分
- Spark2.x 快速入门教程 6
- Spark2.x 快速入门教程 7
- Spark2.x学习笔记:16、Spark Streaming入门实例NetworkWordCount
- Spark基础脚本入门实践3:Pair RDD开发
- spark2.x由浅入深深到底系列六之RDD java api详解三
- spark2.x由浅入深深到底系列六之RDD 支持java8 lambda表达式
- Spark入门——1:RDD及编程接口
- Spark基础入门(二)--------DAG与RDD依赖
- Spark RDD/Core 编程 API入门系列之map、filter、textFile、cache、对Job输出结果进行升和降序、union、groupByKey、join、reduce、look
- Spark入门RDD操作