文章目录
一、前言
二、DStream 编程模型
三、DStream 操作
3.1 套接字流：通过监听 Socket 端口来接收数据
3.2 文件流
3.2 RDD队列流
3.4 带状态的处理 StateFull
一、前言
从前一篇博客 【Spark Streaming】（一）架构及工作原理 🌺，我们了解到 Sprak Streaming 是属于 Saprk API 的扩展，它支持实时数据流（live data streams）的可扩展，高吞吐（hight-throughput） 容错（fault-tolerant）的流处理。可以接受来自KafKa、Flume、ZeroMQ、Kinesis 、Twitter或TCP套接字的数据源，处理的结果数据可以存储到文件系统、数据库、现场dashboards等。
二、DStream 编程模型
Dstream 是 Spark Streaming 中的高级抽象连续数据流，这个数据源可以从外部获得（如KafKa / Flume 等），也可以通过输入流获得，还可以通过在其他 DStream 上进行高级操作创建，DStream 是通过一组时间序列上连续的 RDD表示的，所以一个 DStream 可以看作是一个 RDDs 的序列。（关于 DStream 的深入了解，可看第一篇博客 ！ 🚀）
三、DStream 操作
3.1 套接字流：通过监听 Socket 端口来接收数据
通过Scala编写程序来产生一系列的字符作为输入流：
GenerateChar：
object GenerateChar { def generateContext(index : Int) : String = { import scala.collection.mutable.ListBuffer val charList = ListBuffer[Char]() for(i <- 65 to 90) charList += i.toChar val charArray = charList.toArray charArray(index).toString } def index = { import java.util.Random val rdm = new Random rdm.nextInt(7) } def main(args: Array[String]) { val listener = new ServerSocket(9998) while(true){ val socket = listener.accept() new Thread(){ override def run() = { println("Got client connected from :"+ socket.getInetAddress) val out = new PrintWriter(socket.getOutputStream,true) while(true){ Thread.sleep(500) val context = generateContext(index) //产生的字符是字母表的前七个随机字母 println(context) out.write(context + '\n') out.flush() } socket.close() } }.start() } } }
ScoketStreaming:
object ScoketStreaming { def main(args: Array[String]) { //创建一个本地的StreamingContext，含2个工作线程 val conf = new SparkConf().setMaster("local[2]").setAppName("ScoketStreaming") val sc = new StreamingContext(conf,Seconds(10)) //每隔10秒统计一次字符总数 //创建珍一个DStream，连接master:9998 val lines = sc.socketTextStream("master",9998) val words = lines.flatMap(_.split(" ")) val wordCounts = words.map(x => (x , 1)).reduceByKey(_ + _) wordCounts.print() sc.start() //开始计算 sc.awaitTermination() //通过手动终止计算，否则一直运行下去 } }
运行结果：
GenerateChar 产生的数据如下：
Got client connected from :/192.168.56.137 C G B C F G D G B
ScoketStreaming 运行结果：
------------------------------------------- Time: 1459426750000 ms ------------------------------------------- (B,1) (G,1) (C,1) ------------------------------------------- Time: 1459426760000 ms ------------------------------------------- (B,5) (F,3) (D,4) (G,3) (C,3) (E,1)
注意：如果是在本地运行的，setMaster 的参数必须为local[n]，n >1,官网解释：
When running a Spark Streaming program locally, do not use “local” or “local[1]” as the master URL. Either ofthese means that only one thread will be used for running tasks locally. If you are using a input DStream based on a receiver (e.g. sockets, Kafka, Flume, etc.), then the single thread will be used to run the receiver,leaving no thread for processing the received data. 当在本地运行Spark Streaming程序时,Master的URL不能设置为"local"或"local[1]",这两种设置都意味着你将会只有一个线程来运行作业，如果你的Input DStream基于一个接收器 （如Kafka，Flume等），那么只有一个线程来接收数据，而没有多余的线程来处理接收到的数据。
如果是在集群上运行，为 Spark streaming 应分配的核数应该在大于接收器的数据，否则同样只接收了数据而没有能力处理。
3.2 文件流
Spark Streaming 通过监控文件系统的变化，若有新文件添加，则将它读入并作为数据流
需要注意的是：
这些文件具有相同的格式
这些文件通过原子移动或重命名文件的方式在dataDirectory创建
一旦移动这些文件，就不能再进行修改，如果在文件中追加内容，这些追加的新数据也不会被读取。
FileStreaming：
object FileStreaming { def main(args: Array[String]) { val conf = new SparkConf().setMaster("local").setAppName("FileStreaming") val sc = new StreamingContext(conf,Seconds(5)) val lines = sc.textFileStream("/home/hadoop/wordCount") val words = lines.flatMap(_.split(" ")) val wordCounts = words.map(x => (x , 1)).reduceByKey(_ + _) sc.start() sc.awaitTermination() } }
当你在文件目录里添加文件时，Spark Streaming就会自动帮你读入并计算 ，可以读取本地目录 HDFS和其他文件系统。
注意：文件流不需要运行接收器，所以不需要分配核数
3.2 RDD队列流
使用 StreamingContext.queueStream(queueOfRDD) 创建基于 RDD队列 的 DStream ，用于调试 Spark Streaming 应用程序。 QueueStream：程序每隔1秒就创建一个RDD，Streaming每隔1秒就就对数据进行处理。
object QueueStream { def main(args: Array[String]) { val conf = new SparkConf().setMaster("local[2]").setAppName("queueStream") //每1秒对数据进行处理 val ssc = new StreamingContext(conf,Seconds(1)) //创建一个能够push到QueueInputDStream的RDDs队列 val rddQueue = new mutable.SynchronizedQueue[RDD[Int]]() //基于一个RDD队列创建一个输入源 val inputStream = ssc.queueStream(rddQueue) val mappedStream = inputStream.map(x => (x % 10,1)) val reduceStream = mappedStream.reduceByKey(_ + _) reduceStream.print ssc.start() for(i <- 1 to 30){ rddQueue += ssc.sparkContext.makeRDD(1 to 100, 2) //创建RDD，并分配两个核数 Thread.sleep(1000) } ssc.stop() } }
输出：
------------------------------------------- Time: 1459595433000 ms //第1个输出 ------------------------------------------- (4,10) (0,10) (6,10) (8,10) (2,10) (1,10) (3,10) (7,10) (9,10) (5,10) ............ ............ ------------------------------------------- Time: 1459595463000 ms //第30个输出 ------------------------------------------- (4,10) (0,10) (6,10) (8,10) (2,10) (1,10) (3,10) (7,10) (9,10) (5,10)
3.4 带状态的处理 StateFull
updateStateByKey 操作：使用updateStateByKey操作的地是为了保留key的状态，并能持续的更新；使用此功能有如下两个步骤：
- 定义状态，这个状态可以是任意的数据类型
- 定义状态更新函数， 指定一个函数根据之前的状态来确定如何更新状态。
同样以 wordCount 作为例子，不同的是每一次的输出都会累计之前的 wordCount
StateFull：
object StateFull { def main(args: Array[String]) { //定义状态更新函数 val updateFunc = (values: Seq[Int], state: Option[Int]) => { val currentCount = values.foldLeft(0)(_ + _) val previousCount = state.getOrElse(0) Some(currentCount + previousCount) } val conf = new SparkConf().setMaster("local[2]").setAppName("stateFull") val sc = new StreamingContext(conf, Seconds(5)) sc.checkpoint(".") //设置检查点，存储位置是当前目录，检查点具有容错机制 val lines = sc.socketTextStream("master", 9999) val words = lines.flatMap(_.split(" ")) val wordDstream = words.map(x => (x, 1)) val stateDstream = wordDstream.updateStateByKey[Int](updateFunc) stateDstream.print() sc.start() sc.awaitTermination() } }
先运行之前 GenerateChar 来产生字母，再运行 StateFull，结果如下：
------------------------------------------- Time: 1459597690000 ms ------------------------------------------- (B,3) (F,1) (D,1) (G,1) (C,1) ------------------------------------------- Time: 1459597700000 ms //会累计之前的值 ------------------------------------------- (B,5) (F,3) (D,4) (G,4) (A,2) (E,5) (C,4)
