Spark（十） -- Spark Streaming API编程

本文测试的Spark版本是1.3.1

Spark Streaming编程模型：

第一步：
需要一个StreamingContext对象，该对象是Spark Streaming操作的入口 ，而构建一个StreamingContext对象需要两个参数：
1、SparkConf对象：该对象是配置Spark 程序设置的，例如集群的Master节点，程序名等信息
2、Seconds对象：该对象设置了StreamingContext多久读取一次数据流

第二步：
构建好入口对象之后，直接调用该入口的方法读取各种不同方式传输过来的数据流，如：Socket，HDFS等方式。并会将数据转换成DStream对象进行统一操作

第三步：
DStream本身是一种RDD序列，Streaming接受数据流之后会进行切片，每个片都是一个RDD，而这些RDD最后都会包装到一个DStream对象中统一操作。在这个步骤中，进行对数据的业务处理

第四步：
调用入口对象的start和awaitTermination开始读取数据流

下面分别使用不同的Spark Streaming 处理方式完成WordCount单词计数

HDFS文件测试

object HDFSWordCount { def main(args: Array[String]) { //参数设置 if (args.length < 2) { System.err.println("Usgae : <spark master> <hdfs path>") System.exit(1) } //第一步：创建StreamingContext入口 val sparkConf = new SparkConf().setMaster(args(0)).setAppName("HDFSWordCount") val streaming = new StreamingContext(sparkConf,Seconds(10)) //第二步：调用textFileStream读取指定路径的文件 val data = streaming.textFileStream(args(1)) //第三步，数据业务处理 //使用flatMap将数据map之后的分切压成一个DStream val words = data.flatMap(_.split(" ")) val wordCount = words.map(x => (x,1)).reduceByKey(_+_) wordCount.print() //第四步 streaming.start() streaming.awaitTermination() }

Socket数据流测试

object NetworkWordCount { def main(args: Array[String]) { if (args.length < 3) { System.err.println("Usage : <spark master> <hostname> <port>") System.exit(1) } val sparkConf = new SparkConf().setMaster(args(0)).setAppName("NetworkWordCount") val streaming = new StreamingContext(sparkConf,Seconds(10)) //参数：1、主机名；2、端口号；3、存储级别 val data = streaming.socketTextStream(args(1),args(2).toInt,StorageLevel.MEMORY_AND_DISK_SER) val words = lines.flatMap(_.split(" ")) val wordCount = words.map(x => (x, 1)).reduceByKey(_ + _) wordCount.print() streaming.start() streaming.awaitTermination() }

可以看到，对于同一中业务处理逻辑来说，不同的数据来源只要调用不同的方法接收即可，转换成DStream之后的处理步骤是一模一样的

下面的代码时配合测试Socket数据的，使用java命令执行jar包，传入参数：1、端口号；2、产生数据的频率（毫秒）
即可在指定的端口上产生数据提供Spark Streaming接收

package Streaming import java.net.ServerSocket import java.io.PrintWriter object Logger { def generateContent(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 ran = new Random ran.nextInt(7) } def main(args: Array[String]): Unit = { if (args.length != 2) { System.err.println("Usage:<port> <millisecond>") System.exit(1) } val listener = new ServerSocket(args(0).toInt) while (true) { val socket = listener.accept() new Thread() { override def run = { println("Get client connected from:" + socket.getInetAddress) val out = new PrintWriter(socket.getOutputStream(), true) while (true) { Thread.sleep(args(1).toLong) val content = generateContent(index) println(content) out.write(content + '\n') out.flush() } socket.close() } }.start() } } }

在上述的例子中，文中使用的是Seconds（10）
也就是说每10秒钟处理一次数据
第一个10秒处理的结果是不会影响到第二个10秒的
但是有时候我们需要进行汇通统计，要用到之前几个10秒阶段的数据怎么办？

这里要用到一个updateStateByKey方法，该方法会保存上次计算数据的状态，以供下次计算使用。
上代码：

object StatefulWordCount { def main(args: Array[String]) { if (args.length < 3) { System.err.println("Usage : <spark master> <hostname> <port>") System.exit(1) } //定义一个匿名函数，并赋值给updateFunc //该函数是updateStateByKey方法的参数，该方法要求传入一个匿名参数且参数格式为values:Seq[Int],state:Option[Int] //其中values是当前的数据，state表示之前的数据 //这个匿名函数的作用就是将各个10秒阶段的结果累加汇总 val updateFunc = (values:Seq[Int],state:Option[Int]) => { val now = values.foldLeft(0)(_+_) val old = state.getOrElse(0) Some(now + old) } val conf = new SparkConf().setAppName("StatefulWordCount").setMaster(args(0)) val streaming = new StreamingContext(conf, Seconds(10)) //checkpoint会将数据放在指定的路径上，这个操作是必须的，为了保护数据，如果不设置会报异常 streaming.checkpoint(".") val lines = streaming.socketTextStream(args(1), args(2).toInt, StorageLevel.MEMORY_AND_DISK_SER) val words = lines.flatMap(_.split(" ")) val wordDStream = words.map(x => (x, 1)) //在这里将updateFunc传入 val stateDStream = wordDStream.updateStateByKey(updateFunc) stateDStream.print() streaming.start() streaming.awaitTermination() }

在Spark Streaming中还有一个window的概念，即滑动窗体

下图是官方文档中给出的解释：

使用滑动窗体要设置两个指定参数：
1、窗体长度
2、滑动时间
例如，设置一个窗体长度为5，滑动时间为2，意味着，每2秒处理上一个5秒内的数据流
这样的处理可以应用在例如微博统计最热搜索词
每2秒钟统计一次过去5秒内的最热搜索词

统计最热搜索词实例代码：

object WindowWordCount { def main(args: Array[String]) { if (args.length < 3) { System.err.println("Usage : <spark master> <hostname> <port> <Streaming Seconds> <Window Seconds> <Slide Seconds>") System.exit(1) } val conf = new SparkConf().setAppName("WindowWordCount").setMaster(args(0)) val streaming = new StreamingContext(conf, Seconds(args(3).toInt)) //checkpoint会将数据放在指定的路径上，这个操作是必须的，为了保护数据，如果不设置会报异常 streaming.checkpoint(".") val lines = streaming.socketTextStream(args(1), args(2).toInt, StorageLevel.MEMORY_ONLY) val words = lines.flatMap(_.split(" ")) //map操作之后数据的格式为： //(a,1)(b,1)...(n,1)格式 //调用reduceByKeyAndWindow替代普通的reduceByKey //最后两个参数分别是窗体长度和滑动时间 val wordCount = words.map(x => (x, 1)).reduceByKeyAndWindow(_ + _, _ - _, Seconds(args(4).toInt), Seconds(args(5).toInt)) //对结果进行降序排序 //由于DStream本身不具备RDD的一些操作，调用transform方法可以让RDD的一些操作（例如sortByKey等）作用在其之上，返回的仍然是一个DStream对象 val sorted = wordCount.map { case (char, count) => (count, char) }.transform(_.sortByKey(false)).map { case (count, char) => (char, count) } sorted.print() streaming.start() streaming.awaitTermination() } }

reduceByKeyAndWindow有两种使用方法：
1、educeByKeyAndWindow(_ + _, Seconds(5)，seconds(1))
2、reduceByKeyAndWindow(_ + , - _, Seconds(5)，seconds(1))

二者的区别见下图：

第一种是简单粗暴的直接累加
而第二种方式就显得比较文雅和高效了
例如现在计算t+4的累积数据
第一种方式是，直接从t+…+（t+4）
第二种处理为，用已经计算好的（t+3）的数据加上（t+4）的数据，在减去（t-1）的数据，就可以得到和第一种方式一样的结果，但是中间复用了三个数据（t+1，t+2，t+3）

以上为Spark Streaming API的简单使用