Spark Streaming
一、大数据实时计算
1.实时计算
Spark Streaming ,其实就是一种spark提供的，对于大数据，进行实时计算的一种框架。他的底层，其实也是基于我们之前讲解的Spark core的。基本的计算模型，还是基于内存的大数据实时计算模型。而且，他的底层的组件或者叫做概念，其实还是最核心的RDD
只不过。针对实时计算的特点，在RDD之上，进行了一层封装，叫做Dstream。其实，了解spark sql之后，你理解这种封装就容易了，sparksql针对数据查询，提供了一种基于RDD之上的全新的概念,DataFrame，但是，其底层还是基于RDD的，所以，RDD是整个spark技术生态中的核心。要学好spark在交互式查询、实时计算上的应用技术和框架，首先要耴spark核心编程，也就是spark core
二、Spark Streaming 基本工作原理
1.工作原理
2.Spark Streaming 简介
Spark Streaming是Spark Core API 的一种扩展，他可以用于进行大规模、高吞吐、容错的实时数据流的处理，它支持很多数据源中读取数据，比如kafka,flume,Twitter,ZeroMQ,kinesis或者是TCP Socket。并且能够使用类似高阶函数的复杂算法来进行数据处理，比如map reduce join window 。处理后的数据可以被保存到文件系统、数据库、Dashboard等存储中。
3.DStream
Spark Streaming提供了一种高级的抽象，叫做DStream.英文全称叫做Discretized Stream,中文翻译为“离散流”，它代表了一个持续不断地数据流。DStream可以通过输入数据源来创建，比如kafka,flume,和kinesis；也可以通过对其他DStream应用高阶函数来创建，比如map reduce join window。DStream的内部，其实一系列持续不断的产生RDD，RDD是spark core的核心抽象，即不可变得，分布式的数据集，DStream中的每个RDD都包含了一个时间段的数据。
对于DStream应用的算子，比如map 其实底层会被翻译成对DStream中的每一RDD的操作。比如对一个DStream执行一个map操作，会产生一个新的DStream，但是，在底层，其实其原理为，对输入DStream中每个时间段内的RDD，都应用一遍map操作，然后会产生新的RDD，即作为新的DStream中的那个时间段的一个RDD，底层的RDD的transformation操作，其实还是由spark core的计算引擎来实现的。Spark streaming对spark core进行了一层封装，隐藏了细节，然后对开发人员提供了方便易用的高层次的API
4.Spark Streaming 与Storm对比
|
对比点 |
Storm |
Spark Streaming |
|
实时计算模型 |
纯实时，来一条数据，处理一条数据 |
准实时，对一个时间段内的数据收集起来，作为一个RDD，再进行处理 |
|
实时计算延迟度 |
毫秒级 |
秒级 |
|
吞吐量 |
低 |
高 |
|
事务机制 |
支持完整 |
支持，但是不够完善 |
|
健壮性/容错性 |
Zooleeper,Acker,非常强 |
Checkpoint,WAL，一般 |
|
动态调整并行度 |
支持 |
不支持 |
Spark Streaming 与Storm 的优劣分析
事实上，Spark Streaming绝对谈不上比Storm优秀，这两个框架在实时计算领域，都很优秀，只是擅长的细分场景并不相同
Spark Streaming仅仅在吞吐量上比Storm要优秀，而吞吐量这一点，也是历年来支持Spark Streaming的人着重强调的，但是问题是，是不是所有的实时计算场景下，都那么注重吞吐量？不尽然，因为，通过吞吐量说Spark Streaming 强于Storm ，不靠谱
事实上，Storm在实时延迟上，，比Spark Streaming就好多了，前者是纯实时，后者是准实时。而且。Storm的事务机制、健壮性/容错性、动态调整并行度等特性，都要比Spark Streaming更加优秀
Spark Streaming，有一点是Storm绝对比不上，就是，它位于Spark生态技术栈，因此Spark Streaming可以和Spark Core、Spark Sql 无缝整合，也就意味着，我们可以对实时处理出来的中间数据，立即在程序中无缝进行延迟批处理、交互式查询等操作。这个特点大大增加了Spark Streaming的优势和功能。
Spark Streaming与Storm的应用场景
对于Storm来说：
1.建议在那种需求纯实时，不能忍受1秒以上的延迟的场景下使用，比如实时金融系统，要求纯实时进行金融交易和分析
2.此外，如果对于实时计算的功能中，要求可靠的事务机制和可靠性机制，即数据的处理完全准确，一条也不能多，一条也不能少，也可以考虑使用storm
3.如果还需要针对高峰低峰时间段，动态调整实时计算程序的并行度，以最大限度利用集群资源（通常是在小型公司，集群资源比较紧张的情况），也可以考虑用storm
4.如果一个大数据应用系统，他就是纯粹的实时计算，不需要在中间执行sql交互式查询、复杂的transformation算子等，那么就用storm是比较好的选择
对于spark streaming来说：
1.如果对上述适用于storm的三点，一条都不满足的实时场景下，也就是不要求纯实时，不要求强大的可靠的事务机制，不要求动态的调整并行度，那么可以考虑使用Spark Streaming
2.考虑使用Spark Streaming最主要的一个因素，应该是针对整个项目进行宏观的考虑，即，如果一个项目除了准实时的计算之外，还包括离线处理，交互式查询等功能，而且事实计算中，可能还会牵扯到高延迟批处理、交互式查询等功能，那么首先Spark生态，用Spark Core开发离线批处理，用Spark Sql开发交互式查询，用Spark Streaming开发实时计算，三者可以无缝的整合，给系统提供非常高的可扩展性
三、实时WordCount程序开发
1.安装nc工具，yum install nc
nc -lk 9999
2.开发实时wordcount
Java版本：
|
package com.spark.spark_streaming; import java.util.Arrays; import org.apache.spark.SparkConf; import org.apache.spark.api.java.function.FlatMapFunction; import org.apache.spark.api.java.function.Function2; import org.apache.spark.api.java.function.PairFunction; import org.apache.spark.streaming.Durations; import org.apache.spark.streaming.api.java.JavaDStream; import org.apache.spark.streaming.api.java.JavaPairDStream; import org.apache.spark.streaming.api.java.JavaReceiverInputDStream; import org.apache.spark.streaming.api.java.JavaStreamingContext; import scala.Tuple2; /** * 实时wordcount程序 */ public class WordCount { public static void main(String[] args) { //创建一个sparkconf对象 //但是这有一点不同，我们是要给他设置一个master属性，但是我们测试的时候使用local模式 //local后面必须跟一个方括号，里面填一个数子，数字代表了，我们用几个线程来执行我们的spark streaming程序 SparkConf conf = new SparkConf().setMaster("local[2]").setAppName("WordCount"); //创建一javastreamingContext对象 //该对象，就类似与spark core的javasparkcontext/ spark sql中的sqlcontext //该对象处理接收一个batch interval 参数之外 //还必须接收一个batch interval参数，就是说，每收集多长时间的数据，划分为一个batch 进行处理 //这里设置为10秒 JavaStreamingContext jssc = new JavaStreamingContext(conf,Durations.seconds(10)); //首先，创建输入dstream,代表了一个从数据源（kafkasocket）来的持续不断的实时数据流 //数据流，JavaReceiverInputDStream代表了一个输入的dstream JavaReceiverInputDStream<String> lines= jssc.socketTextStream("localhost", 9999); //JavaReceiverInputDStream中每隔10秒就会有一个RDD，其中封装了这10秒发送过来的数据 //RDD的类型是string 即一行一行的文本 //使用spark core提供的算子，直接引用在dstream上 //在底层实际上会对DStream中的一个一个的RDD执行我们应用在DStream上的算子 //产生新的RDD，会作为新的dstream中的RDD JavaDStream<String> words = lines.flatMap(new FlatMapFunction<String, String>() { private static final long serialVersionUID = 1L; @Override public Iterable<String> call(String line) throws Exception { return Arrays.asList(line.split(" ")); } }); JavaPairDStream<String, Integer> pairs = words.mapToPair(new PairFunction<String, String, Integer>() { private static final long serialVersionUID = 1L; @Override public Tuple2<String, Integer> call(String word) throws Exception { return new Tuple2<String, Integer>(word, 1); } }); JavaPairDStream<String, Integer> wordCounts = pairs.reduceByKey(new Function2<Integer, Integer, Integer>() { @Override public Integer call(Integer v1, Integer v2) throws Exception { return v1 + v2; } }); //每10秒钟发送到指定的端口上的数据，都会被lines DStream接收到 //然后lines DStream会把每10秒的数据也就是一行一行的文本，诸如hello world,封装成一个RDD //然后就会对每10秒对应的RDD执行一系列的算子操作 //比如对lines RDD执行flatmap之后，得到words RDD 作为words Dstream的一个RDD //以此类推，直到生成最后一个wordcount RDD，作为wordcounts DStream中的一个RDD //此时就到了每10秒钟发过来的数据进行单词统计 ///但是一定要注意，Spark Streaming的计算模型，就决定了，我们必自己来进行中间缓存的控制 //比如写到redis等缓存 //他的计算模型跟Storm是完全不同的，Storm是自己编写的一个一个的程序，运行在节点上 //相当于一个一个的对象，可以自己在对象中控制缓存 //但是spark本身是函数式编程的计算模型，所以，比如在words或pairs Dstream中没法在实例变量中进行缓存 //此时，就只能将最后的计算出的wordcounts中一个一个的RDD，写入外部的缓存，或者持久化到DB wordCounts.print(); //对JavaStreamingContext做后续的处理 //必须调用JavaStreamingContext的start()方法，整个spark streaming application才会启动执行 //否则是不会执行的 jssc.start(); jssc.awaitTermination(); jssc.close(); } } |
Scala版本：
|
package com.spark.spark_streaming import org.apache.spark.SparkConf import org.apache.spark.streaming.StreamingContext import org.apache.spark.sql.catalyst.expressions.Second import org.apache.spark.streaming.Seconds object WordCount { def main(args: Array[String]): Unit = { val conf = new SparkConf().setMaster("local[2]").setAppName("WordCount") val ssc = new StreamingContext(conf,Seconds(10)); val lines = ssc.socketTextStream("localhost", 9999) val words = lines.flatMap(line => line.split(" ")) val pairs = words.map(word => (word , 1)) val wordCounts = pairs.reduceByKey(_+_) wordCounts.print() ssc.start() ssc.awaitTermination() } } |
四、StreamingContext详解
1.创建StreamingContext
有两种创建StreamingContext的方式：
Val conf = new sparkConf().setAppName(appName).setMaster(master)
Val ssc = new StreamingContext(conf,Seconds(1))
StreamingContext，还可以使用已有的sparkContext来创建
Val sc = new sparkContext(conf)
Val ssc = new StreamingContext(sc,seconds(1));
appName，使用来在spark UI 上显示的应用的名称。Master是一个spark、mesos、或者yarn 集群的url或者是local[*]
Batch interval可以根据你的应用程序的延迟要求以及可用的集群资源情况来进行设置。
2.要点：
一个StreamingContext定义之后，必须做以下几件事：
1.通过创建输入源数据来创建DStream
2.通过对DStream定义transformation和output算子操作，来定义实时计算逻辑
3.调用StreamingContext的start()方法，来开始实时处理数据
4.调用StreamingContext的awaitTemination()方法，来等待应用程序的终止。可以使用CTRL+C手动停止，或者就是让他持续不断的运行计算
5.也可以通过调用StreamingContext的stop方式，来停止应用程序
需要注意的要点：
1.只要一个StreamingContext启动之后，就不能再往其中添加任何计算逻辑了，比如执行start方法之后，还给某个DStream执行一个算子
2.一个StreamingContext停止之后，是肯定不能重启的。调用stop之后，不能再调用start
3.一个JVM同时只能有一个StreamingContext启动。在你的应用程序中，不能创建两个StreamingContext。
4.调用stop()方法时，会同时停止内部的SparkContext,如果不希望如此，还希望后面继续使用SparkContext创建其他类型的Context，比如SQLContext,那么就用stop（false）
5.一个sparkContext可以创建多个StreamingContext，只要把上一个先用stop（false）停止，在创建下一个即可。
五、输入DStream和Receiver
1.详解
输入DStream代表了来自数据源的数据流。在之前的wordcount例子中，lines就是一个输入DStream，（javaRecieverInputDStream）,代表了从netcat（nc）服务接收到的数据流。除了关键的数据流之外，所有的数据DStream都会绑定一个Receiver对象，该对象是一个关键的组件，用来从数据源接收数据，并将其存储到spark内存中
Spark Streaming提供了两种内置的数据源支持：
基础数据源：StreamingContext API中直接提供了对这些数据源的支持，比如文件，socket，AKKa Actor等
高级数据源：kafka、flume、kinesis、Twitter等数据源，通过第三方的工具类提供支持。这些数据源的使用，需要引用其依赖。
要注意的是，如果你想要在实时计算应用中并行接收多条数据流，可以创建多个DStream。这样就会创建多个Receiver,从而并行的接收多个数据源。但是，要注意的是一个spark streaming application的executor,是一个长时间运行的任务，因此他会独占分配给spark streaming application的CPU core。从而只要spark streaming 运行起来以后，这个节点上的cpu core，就没法给其他应用使用了
使用本地模式，运行程序时，绝对不能用了local或者local[1];因为那样的话，只会给执行输入DStream的executor分配一个线程，而Spark Streaming底层原理是，至少要有两个线程，一条线程用来分配给Receiver接收数据，一条线程用来处理接收到的数据。因此，必须使用了local[n],n>=2的模式
如果不设置master ，也就是直接将Spark Streaming应用提交到集群上运行，那么首先，必须要求集群节点上，有大于一个CPU core，其次是给Spark Streaming的每一个executor分配的core，必须大于1，这样，才能保证分配到的executor上运行的输入DStream，两条线程并行，一条运行Receive,接收数据，一条处理数据，否则的话，只会接收数据，不会处理数据
因此，基于此，实际生产环境，机器肯定是不只是一CPU core的，现在至少4核，到时候记得给每个executor的CPU core，设置为超过一个即可
六、输入DStream之基础数据源
1.socket：
StreamingContext.socketTextStream()
2.HDFS文件
基于hdfs文件的实时计算，其实就是，监控一个hdfs目录，只要其中有新文件出现，就实时处理，相当于处理实时的文件流
streamingContext.fileStream<KeyClass,ValueClass,InputFormatClass>(dataDirectory)
streamingContext.fileStream[KeyClass,ValueClass,InputFormatClass](dataDirectory)
Spark streaming会监视指定的hdfs目录，并且处理出现在目录中的文件，要注意的是，所有放入hdfs目录中的文件，都必须有相同的格式，必须使用移动或者重命名的方式，将文件移入目录；一旦处理之后，文件的内容即使改变，也不会在处理了
基于hdfs文件的数据源是没有receiver的，因此不会占用一个cpu core
案例：基于hdfs上的wordcount程序
Java版本：
|
package com.spark.spark_streaming; import java.util.Arrays; import org.apache.spark.SparkConf; import org.apache.spark.api.java.function.FlatMapFunction; import org.apache.spark.api.java.function.Function2; import org.apache.spark.api.java.function.PairFunction; import org.apache.spark.streaming.Durations; import org.apache.spark.streaming.api.java.JavaDStream; import org.apache.spark.streaming.api.java.JavaPairDStream; import org.apache.spark.streaming.api.java.JavaStreamingContext; import scala.Tuple2; public class HDFSWordCount { public static void main(String[] args) { SparkConf conf = new SparkConf().setAppName("HDFSWordCount").setMaster("local[2]"); JavaStreamingContext jsc = new JavaStreamingContext(conf,Durations.seconds(5)); JavaDStream<String> lines = jsc.textFileStream("hdfs://hadoop01:8020/spark_streaming/"); JavaDStream<String> words = lines.flatMap(new FlatMapFunction<String, String>() { private static final long serialVersionUID = 1L; @Override public Iterable<String> call(String line) throws Exception { return Arrays.asList(line.split(" ")); } }); JavaPairDStream<String, Integer> pairs = words.mapToPair(new PairFunction<String, String, Integer>() { private static final long serialVersionUID = 1L; @Override public Tuple2<String, Integer> call(String word) throws Exception { return new Tuple2<String, Integer>(word, 1); } }); JavaPairDStream<String, Integer> wordcount = pairs.reduceByKey(new Function2<Integer, Integer, Integer>() { private static final long serialVersionUID = 1L; @Override public Integer call(Integer v1, Integer v2) throws Exception { return v1 + v2; } }); wordcount.print(); jsc.start(); jsc.awaitTermination(); jsc.close(); } } |
Scala版本：
|
package com.spark.spark_streaming import org.apache.spark.SparkConf import org.apache.spark.streaming.StreamingContext import org.apache.spark.streaming.Seconds object HDFSWordCount { val conf = new SparkConf().setMaster("local[2]").setAppName("HDFSWordCount") val ssc = new StreamingContext(conf,Seconds(5)) val lines = ssc.textFileStream("hdfs://hadoop01:8020/spark_streaming") val words = lines.flatMap(line => line.split(" ")) val pairs = words.map(word => (word,1)) val count = pairs.reduceByKey(_+_) count.print(); ssc.start() ssc.awaitTermination(); } |
七、kafka数据源
1. 基于Receiver的方式
这种方式使用Receiver来获取数据。Receiver是使用kafka的高层次Consumer API来实现的。Receiver从kafka中获取的数据都是存储在Spark executor的内存中的，然后将Spark Streaming启动的job会去处理那些数据
然而，在默认的情况下，这种方式可能会因为底层的失败而丢失数据。如果要启动高可靠的机制，让数据零丢失，就必须启用Spark Streaming的预写日志的（WAL）.该机制会同步的将接收到的kafka数据写入分布式文件系统（比如HDFS）上的预写日志中。所以，即使底层出现了失败，也可以使用预写日志中的数据进行恢复
2.如何进行kafka数据源连接
在maven中添加依赖：
Groupid = org.apache.spark
Artifactid=spark-streaming-kafka_2.10
Version= 1.6.1
使用第三方的工具创建输入DStream
javapairReceiverInputDStream<String,String> kafkaStream = kafkaUtils.createStream(streamingContext[ZK quorun,][consumer group id],[per-topic number of kafka partitions to consumer]);
Kafka命令：
Bin/kafka-topic.sh --zookeeper 192,168.8.91:2181 --topic TestTopic --replication-factor 1 --partitions 1 --create
Bin/kafka --console-producer.sh --broker-list 192,168.8.91:9092 --topic TestTopic
需要注意的要点：
1.kafka中的topic的partition，与Spark中的RDD的partition是没有关系的。所以在kafkaUtils.createStream()中，提高partition的数量，只会增加一个Revceiver中读取partition的线程的数量。不会增加spark处理数据的并行度
2.可以创建多个kafka输入DStream，使用不同的consumer group和topic,来通过多个receiver并行接收数据。
3.如果基于容错的文件系统，比如HDFS,启动了预写日志的机制，接收到的数据都会被复制一份到预写日志中。因此kafkaUtils.createStream()中，设置的持久化级别是StorageLevel.MEMORY_AND_DISK_SER
3.基于Direct的方式
这种新的不基于Receiver的直接方式，是在spark1.3中引入的，从而能够确保更加健壮的机制。替代掉使用 Receiver来接收数据后，这种方式会周期性的查询kafka，来获取每一个topic_partition的最新的offset，从而定义每个batch的offset范围。当处理数据的job启动时就会使用kakfa的简单consumerAPI来获取kafka指定offser范围的数据
这种方式的优点：
1、简化并行读取：
如果要读取多个partition，不需要创建多个输入Dstream然后对他们进行union操作。Spark会创建跟kafka partition一样多的RDD partition，并且会并行从kafka中读取数据。所以在kafka partition和RDD partition之间，有一个一对一的映射关系
2、高性能：
如果要保证零数据丢失，在基于receiver的方式中，需要开始WAL机制。这种方式其实效率很低，因为数据实际上是被复制两份，kafka自身就有高可靠机制。会对数据复制一份，而这里又会复制一个到WAL中。而基于direct的方式，不依赖Receiver，不需要开启WAL机制，只要kafka中做了数据的复制，那么就可以通过kafka的副本进行恢复
3.一次且仅一次的事务机制
基于receiver的方式，是使用kafka的高阶API在zookeeper中 保存消费过的offset的 ，这是消费kafka数据的传统方式。这种方式配合者WAL机制可以保证数据零丢失的高可靠性，但是无法保证数据被处理一次且仅一次，可能会处理两次 ，因为spark和zookeeper可能是不同步的。
基于direct的方式，使用kafka的简单API，Spark Streaming自己就负责追踪消费的offset ，并保存在checkpoint中。Spark自己一定是同步的，因此 可以保证元数据消费一次且仅消费一次
javaPairReceiverInputDStream<String,String> directKafkaStream=KafkaUtils.createDirectStream(streamingContext,[key class],[value class],[key decoder class],[value decoder class],[map of kafka parameters],[set of topics to consume]);
八、DStream的transformation操作
1.概述
|
transformation |
meaning |
|
map |
对传入的每一个元素，返回一个新的元素 |
|
flatMap |
对传入的每一个元素，返回一个或多个元素 |
|
filter |
对传入的元素返回true或false，返回true的元素被保留，返回false的元素被过滤 |
|
union |
将两个DStream进行合并 |
|
count |
返回元素的个数 |
|
reduce |
对所有的values进行聚合 |
|
countByValue |
对元素按照值进行分组，对每个组进行计数，最后返回<k,v>的格式 |
|
reduceByKey |
对key对应的value进行聚合 |
|
cogroup |
对两个DStream进行连接操作，一个key连接起来的两个RDD的数据，都会以Iterable<V>的形式，出现在一个Tuple中 |
|
join |
对两个DStream进行join操作，每个连接起来的pair，最为新的DStream的RDD的一个元素 |
|
transform |
对数据进行转换操作 |
|
UpdateStateByKey |
为每一个key维护一份state，并进行更新（普通的实时计算中最有用的一种操作） |
|
window |
对滑动窗口数据执行操作（实时计算中最有特色的一种操作） |
【注意】 比如有一个RDD（1,1）（1,2）（1,3）另外一个RDD（1,4）（2,1）（2,2）
进行join：（1，（1,4））（1，（2,4））（1，（3,4））（）（）
进行cogroup：（1，（1,2,3），（4））
2.updateStateByKey
updateStateByKey操作，可以让我们为每一个key维护一份state，并持续不断的更新该key
首先，要定义一个state，可以是任意的数据类型
其次，要定义state更新函数：指定一个函数如何使用之前的state和新值来更新state
对于每个batch，spark都会为每个之前已经存在的key去应用一次state更新函数，无论这个key在batch中是否有新的数据。如果state更新函数返回none,那么key对应的state就会被删除
当然，对于每个新出现的key，也会执行state更新函数
注意，updateStateByKey操作，要求必须开启Checkpoint机制
案例：基于缓存的实时wordcount程序（实际的业务场景中是非常有用的）
Java版本
|
package com.spark.spark_streaming; import java.util.Arrays; import java.util.List; import org.apache.spark.SparkConf; import org.apache.spark.api.java.function.FlatMapFunction; import org.apache.spark.api.java.function.Function2; import org.apache.spark.api.java.function.PairFunction; import org.apache.spark.streaming.Durations; import org.apache.spark.streaming.api.java.JavaDStream; import org.apache.spark.streaming.api.java.JavaPairDStream; import org.apache.spark.streaming.api.java.JavaReceiverInputDStream; import org.apache.spark.streaming.api.java.JavaStreamingContext; import com.google.common.base.Optional; import scala.Tuple2; /** * 基于缓存的实时wordcount程序 * @author Administrator * */ public class UpdateStateByKeyWordCount { public static void main(String[] args) { SparkConf conf = new SparkConf().setMaster("local[2]").setAppName("UpdateStateByKeyWordCount"); JavaStreamingContext jsc = new JavaStreamingContext(conf, Durations.seconds(5)); JavaReceiverInputDStream<String> lines = jsc.socketTextStream("localhost", 9999); //第一点：如果要使用updateStateByKey算子，就必须设置checkpoint目录，开启checkpoint用checkpoint的机制 //这样的话，才能把每个key对应的state除了在内存中有，也要checkpoint一份 //因为要长期保存一份key的state的话，那么spark streaming是要求必须用checkpoint的，以便在内存数据丢失的时候， //可以从checkpoint中恢复数据 //开启checkpoint机制，很简单，指点调用jsc的checkpoint方法，设置一个hdfs目录即可 jsc.checkpoint("hdfs://hadoop01:8020/spark_checkpoint"); //然后实现基础的wordcount逻辑 JavaDStream<String> words = lines.flatMap(new FlatMapFunction<String, String>() { private static final long serialVersionUID = 1L; @Override public Iterable<String> call(String line) throws Exception { return Arrays.asList(line.split(" ")); } }); JavaPairDStream<String, Integer> pairs = words.mapToPair(new PairFunction<String, String, Integer>() { private static final long serialVersionUID = 1L; @Override public Tuple2<String, Integer> call(String word) throws Exception { return new Tuple2<String, Integer>(word, 1); } }); //到这里就不一样了，之前的话，就只直接是reduceBykey，然后就可以得到每个时间段batch对应的RDD，计算出单词计数 //但是，有个问题，你如果要统计每个单词的全局计数呢？ //就是说，统计出来，从程序开始带现在为止，一个单词出现的次数，那么就之前的方式不好实现 //就必须基于redis这种缓存，或者是MySQL这种DB，来实现累加 //但是。updateStateByKey,就可以实现直接通过spark维护每个单词的全局统计次数 JavaPairDStream<String, Integer> count = pairs.updateStateByKey(new Function2<List<Integer>, Optional<Integer>, Optional<Integer>>() { //这里的optional相当于scala中的样例类，就是option //它代表了一个值的存在的状态，可能存在，也可能不存在 //这里面两个参数 //实际上，对于每个单词，每次batch计算的时候，都会调用这个函数 //第一个参数：values相当于是这个batch中，这个key的新值，可能有多个 //比如说一个hello 可能有两个1 ,(hello ,1)(hello ,1)那么传入的是(1,1) //第二个参数，就是指的是这个key之前的状态，state，其中泛型类型是你自己执行的 @Override public Optional<Integer> call(List<Integer> values, Optional<Integer> state) throws Exception { //首先定义一个全局的单词计数 Integer newValue=0; //其次，判断state是否存在，如果不存在，说明是一个key第一次出现 //如果存在，说明这个key之前统计过全局的次数了 if(state.isPresent()){ newValue=state.get(); } //接着将本次新出现的值，都累加到newvalue上去，就是一个key目前的全局的统计次数 for (Integer value : values) { newValue +=value; } return Optional.of(newValue); } }); //到这里为止，相当于是每一个batch过来后，计算到pairs DStream.就会执行全局的updateStateByKey算子 //updateStateByKey，返回javapairDStream，其实就是代表每个key的全局的统计。 //打印出来 count.print(); jsc.start(); jsc.awaitTermination(); jsc.close(); } } |
3.transform
Transform操作，应用在DStream上的时候，可以用于执行任意的RDD到RDD的转换操作。他可以用于实现，DStreamAPI中所没有提供的操作。比如说DStreamAPI中，并没有提供将一个DStream中的每个batch，与一个特定的RDD进行操作。但是我们自己就可以使用transform操作来实现该功能
DStream.join()只能join其他DStream。在DStream每个batch的RDD计算出来之后，会去其他的DStream的RDD进行join。
案例：实时黑名单过滤
Java 版本：
|
package com.spark.spark_streaming; import java.util.ArrayList; import java.util.List; import org.apache.spark.SparkConf; import org.apache.spark.api.java.JavaPairRDD; import org.apache.spark.api.java.JavaRDD; import org.apache.spark.api.java.function.Function; import org.apache.spark.api.java.function.PairFunction; import org.apache.spark.streaming.Durations; import org.apache.spark.streaming.api.java.JavaDStream; import org.apache.spark.streaming.api.java.JavaPairDStream; import org.apache.spark.streaming.api.java.JavaReceiverInputDStream; import org.apache.spark.streaming.api.java.JavaStreamingContext; import com.google.common.base.Optional; import scala.Tuple2; /** * 基于transform的实时黑名单过滤 * @author Administrator * */ public class TransformBlacklist { @SuppressWarnings("deprecation") public static void main(String[] args) { SparkConf conf = new SparkConf().setMaster("local[2]").setAppName("TransformBlacklist"); JavaStreamingContext jsc =new JavaStreamingContext(conf, Durations.seconds(5)); //用户对我们的网站上的广告点击，点击之后，进行实时计费 //但是对于那些无良商家刷广告的人，有一个黑名单，只要是黑名单中的用户点击的广告，我们就过滤掉 //这里的日志格式date username的方式 JavaReceiverInputDStream<String> adsClickLogDStream = jsc.socketTextStream("hadoop01", 9999); //模拟黑名单RDD List<Tuple2<String, Boolean>> blacklist = new ArrayList<Tuple2<String,Boolean>>(); blacklist.add(new Tuple2<String, Boolean>("tom", true)); final JavaPairRDD<String, Boolean> blacklistRDD = jsc.sc().parallelizePairs(blacklist); JavaPairDStream<String, String> userAdsClickLogDStream = adsClickLogDStream.mapToPair(new PairFunction<String, String, String>() { private static final long serialVersionUID = 1L; @Override public Tuple2<String, String> call(String adsClickLog) throws Exception { return new Tuple2<String, String>(adsClickLog.split(" ")[1],adsClickLog); } }); //然后，就可以执行join操作了，将每个batch的RDD与黑名单RDD进行join map filter等操作； //实时进行过滤 JavaDStream<String> validAdsClickLogDStream = userAdsClickLogDStream.transform(new Function<JavaPairRDD<String,String>, JavaRDD<String>>() { private static final long serialVersionUID = 1L; @Override public JavaRDD<String> call(JavaPairRDD<String, String> userAdsClickLogRDD) throws Exception { //这为什么用左外连接 //因为并不是每个用户都存在于黑名单中 //所以直接使用join ，那么没有存在于黑名单中的数据会无法join到 //就给丢弃掉了 //所以，这里用leftOuterJoin，就是说哪怕一个user不在黑名单RDD中,没有join到 //也还是会被保存下来的 JavaPairRDD<String, Tuple2<String, Optional<Boolean>>> joinedRDD = userAdsClickLogRDD.leftOuterJoin(blacklistRDD); JavaPairRDD<String, Tuple2<String, Optional<Boolean>>> filteredRDD = joinedRDD.filter(new Function<Tuple2<String,Tuple2<String,Optional<Boolean>>>, Boolean>() { private static final long serialVersionUID = 1L; @Override public Boolean call(Tuple2<String, Tuple2<String, Optional<Boolean>>> tuple) throws Exception { //这里的tuple就是每个用户，对用的访问日志和用户在黑名中的状态 if(tuple._2._2().isPresent() && tuple._2._2().get()){ return false; } return true; } }); //此时。filteredRDD就只剩下没有被黑名单过滤的用户点击了 //进行map操作，装换成我们想要的日志 JavaRDD<String> validAdsClickLogRDD = filteredRDD.map(new Function<Tuple2<String,Tuple2<String,Optional<Boolean>>>, String>() { @Override public String call(Tuple2<String, Tuple2<String, Optional<Boolean>>> tuple) throws Exception { return tuple._2._1; } }); return validAdsClickLogRDD; } }); //有效的广告点击日志 //企业真实的场景是写入kafkaActiveMQ等中间件消息对列 //在开发一个专门的后台服务，最为广告计费服务，执行实时的广告计费，这里就是指导有效的广告点击 validAdsClickLogDStream.print(); //所以，要先对输入的数据，进行下一转换操作，变成（username, date username） //以便于后面的每个batch RDD与定义好的黑名单RDD进行join操作 jsc.start(); jsc.awaitTermination(); jsc.close(); } } |
Scala 版本：
|
package com.spark.spark_streaming import org.apache.spark.SparkConf import org.apache.spark.streaming.StreamingContext import org.apache.spark.streaming.Seconds object TransformBlacklist { def main(args: Array[String]): Unit = { val conf= new SparkConf().setMaster("local[2]").setAppName("TransformBlacklist") val ssc = new StreamingContext(conf,Seconds(10)) val blacklist = Array(("tom",true)) val blacklistRDD = ssc.sparkContext.parallelize(blacklist, 3) val adsClickLogDStream = ssc.socketTextStream("hadoop01", 9999) val userAdsClickLogDStream = adsClickLogDStream.map(adsClickLog =>(adsClickLog.split(" ")(1),adsClickLog)) val validAdsClickLogDStream = userAdsClickLogDStream.transform(userAdsClickLogRDD =>{ val joinedRDD = userAdsClickLogRDD.leftOuterJoin(blacklistRDD) val filteredRDD = joinedRDD.filter(tuple => { if(tuple._2._2.getOrElse(false)){ false }else{ true } }) val validAdsClickLogRDD = filteredRDD.map(tuple => tuple._2._1) validAdsClickLogRDD
}) validAdsClickLogDStream.print
ssc.start() ssc.awaitTermination() } } |
4.window滑动窗口
Spark streaming提供滑动窗口操作的支持，从而让我们可以用一个滑动窗口内的数据进行计算操作。每次掉落到窗口内的RDD的数据，会被聚合起来执行计算操作，然后生成的RDD会作为window DStream的一个RDD，比如下图中，就是对每三秒的数据执行一次滑动窗口的计算，这三秒内的3个RDD会被聚合起来执行处理，然后过了两秒，又会对最近三秒内的数据执行滑动窗口计算。所以每一个滑动窗口操作，都必须执行两个参数，窗口长度以及滑动时间间隔，而且这两个参数的值都必须是batch间隔的整数倍（spark streaming对窗口的支持，是比storm更加完善和强大的）
滑动窗口的操作：
|
transform |
意义 |
|
window |
对每个滑动窗口的数据执行自定义的计算 |
|
countByWindow |
对每个滑动窗口的数据执行count操作 |
|
reduceByWindow |
对每个滑动窗口的数据执行reduce操作 |
|
reduceByKeyAndWindow |
对每个滑动窗口的数据执行reduceByKey操作 |
|
countByValueAndWindow |
对每个滑动窗口的操作执行countByValue操作 |
案例 ：
热点搜索词滑动统计，每隔10秒，统计最近60秒的搜索词的搜索频次，并打印出排名最靠前的3个搜索词以及出现的次数
分析：窗口的长度为60秒，间隔是10秒
Java：
|
package com.spark.spark_streaming; import java.util.List; import org.apache.spark.SparkConf; import org.apache.spark.api.java.JavaPairRDD; import org.apache.spark.api.java.function.Function; import org.apache.spark.api.java.function.Function2; import org.apache.spark.api.java.function.PairFunction; import org.apache.spark.streaming.Durations; import org.apache.spark.streaming.api.java.JavaDStream; import org.apache.spark.streaming.api.java.JavaPairDStream; import org.apache.spark.streaming.api.java.JavaReceiverInputDStream; import org.apache.spark.streaming.api.java.JavaStreamingContext; import scala.Tuple2; /** * 基于滑动窗口热点搜索词实时统计 * @author Administrator * */ public class WindowHotWord { public static void main(String[] args) { SparkConf conf = new SparkConf().setMaster("local[2]").setAppName("WindowHotWord"); JavaStreamingContext jsc = new JavaStreamingContext(conf,Durations.seconds(2)); //说明一下，这里的搜索日志的格式： //leo hello //tom world JavaReceiverInputDStream<String> searchLogDStream = jsc.socketTextStream("hadoop01", 9999); //将搜索日志转为只有一个所搜词即可 JavaDStream<String> searchWordDStream = searchLogDStream.map(new Function<String, String>() { private static final long serialVersionUID = 1L; @Override public String call(String searchLog) throws Exception { return searchLog.split(" ")[1]; } }); //将搜索词映射为(searchWord,1)的tuple的格式 JavaPairDStream<String, Integer> searchWordPairDStream = searchWordDStream.mapToPair(new PairFunction<String, String, Integer>() { private static final long serialVersionUID = 1L; @Override public Tuple2<String, Integer> call(String word) throws Exception { return new Tuple2<String, Integer>(word, 1); } }); //针对(searchword,1)的tuple格式的DStream，执行reduceByKeyAndWindow，滑动窗口操作 //第二个参数是窗口的长度，这里是60秒 //第三个参数是滑动的间隔，这里是10秒 //也就是说每隔10秒钟，将最近60秒的数据作为一个窗口，进行内部的RDD聚合，然后统一对一个RDD进行后溪的计算 //所以说，这里的意思，就是之前的searchWordPairDStream为止，其实都是不会立即执行的 //而是只是放在那里，等待我们的滑动时间到了之后，十秒钟到了，会将之前60秒的RDD，因为一个batch间隔是5秒， //所以之前60秒，就有12个RDD，给聚合起来。然后统一执行reduceByKey, //所以。这里的reduceByKeyAndWindow是针对每个窗口执行计算的，而不是针对某个DStream中国的RDD JavaPairDStream<String, Integer> seachWordCountDStream = searchWordPairDStream.reduceByKeyAndWindow(new Function2<Integer, Integer, Integer>() { private static final long serialVersionUID = 1L; @Override public Integer call(Integer v1, Integer v2) throws Exception { // TODO Auto-generated method stub return v1+v2; } }, Durations.seconds(60),Durations.seconds(10)); //到这里为止，就已经可以做到每隔10秒，出来，将之前60秒的收集到的单词的统计次数 //执行transform操作，因为，一个窗口，就是一个60秒钟中的数据，会变成一个RDD，然后对这一个RDD //根据每个搜索词出现的次数进行排序，然后获得排名前三的热点搜索词 JavaPairDStream<String, Integer> finalDStream = seachWordCountDStream.transformToPair(new Function<JavaPairRDD<String,Integer>, JavaPairRDD<String,Integer>>() { private static final long serialVersionUID = 1L; @Override public JavaPairRDD<String, Integer> call(JavaPairRDD<String, Integer> searchWordCountRDD) throws Exception { //将热点词和出现的频次进行反转 JavaPairRDD<Integer, String> countSearchWordsRDD = searchWordCountRDD.mapToPair(new PairFunction<Tuple2<String,Integer>, Integer, String>() { @Override public Tuple2<Integer, String> call(Tuple2<String, Integer> tuple) throws Exception { return new Tuple2<Integer, String>(tuple._2, tuple._1); } }); //然后执行降序排序 JavaPairRDD<Integer, String> sortedCountSearchWordsRDD = countSearchWordsRDD.sortByKey(false); //再次进行反转，变成（seachWord,count）的这种格式 JavaPairRDD<String, Integer> sortedSearchWordCountRDD = sortedCountSearchWordsRDD.mapToPair(new PairFunction<Tuple2<Integer,String>, String, Integer>() { @Override public Tuple2<String, Integer> call(Tuple2<Integer, String> tuple) throws Exception { return new Tuple2<String, Integer>(tuple._2, tuple._1); } }); //然后用take，获取前三名的热点搜索词 List<Tuple2<String, Integer>> hotSearchWordCounts = sortedSearchWordCountRDD.take(3); for (Tuple2<String, Integer> wordCount : hotSearchWordCounts) { System.out.println(wordCount._1+":"+wordCount._2); } return sortedSearchWordCountRDD; } }); //这个无关紧要，只是为了触发job的执行，所以必须有output操作 finalDStream.print(); jsc.start(); jsc.awaitTermination(); jsc.close(); } } |
Scala：
|
package com.spark.spark_streaming import org.apache.spark.SparkConf import org.apache.spark.streaming.StreamingContext import org.apache.spark.streaming.Seconds object WindowHotWord { def main(args: Array[String]): Unit = { val conf = new SparkConf().setAppName("WindowHotWord").setMaster("local[2]") val ssc = new StreamingContext(conf,Seconds(5)) val seachLogDStream = ssc.socketTextStream("hadoop01", 9999) val seachWordsDStream = seachLogDStream.map(_.split(" ")(1)) val seachWordsPairsDStream = seachWordsDStream.map(searchWord =>(searchWord,1)) val seachWordCountsDStream=seachWordsPairsDStream.reduceByKeyAndWindow( (v1:Int,v2:Int) => v1 + v2,Seconds(60), Seconds(10)) val finalDStream = seachWordCountsDStream.transform(searchWordCountsRDD =>{ val countSeachWordsRDD = searchWordCountsRDD.map(tuple => (tuple._2,tuple._1)) val sortedCountSeachWordsRDD =countSeachWordsRDD.sortByKey(false) val sortedWordCountsRDD = sortedCountSeachWordsRDD.map(tuple => (tuple._2,tuple._1)) val top3SearchWordCountsRDD = sortedWordCountsRDD.take(3)
for(tuple <-top3SearchWordCountsRDD ){ println(tuple) } searchWordCountsRDD }) finalDStream.print()
ssc.start() ssc.awaitTermination() } } |
5.output操作、foreachRDD详解
1.output操作
|
Output |
意义 |
|
|
打印每个batch中的前10个元素，主要用于测试，或者是不需要执行什么output操作时，用于简单的触发一下job |
|
saveAsTextFile(prefix,[suffix]) |
将每个batch的数据保存到文件中。每个batch的文件命名格式是prefix-time_in_mx[.suffix] |
|
saveAsObjectFile |
同上，但是将每个batch的数据一序列化对象的方式，保存到sequenceFile中 |
|
saveAsHadoopFile |
同上，将数据保存到hadoop文件中 |
|
foreachRDD |
最常用的output操作，遍历DStream中的每一个产生的RDD，进行处理，可以将每一个RDD中的数据写入到外部存储，比如文件，数据库，缓存等。通常在其中是针对RDD执行action操作的，比如foreach |
DStream中所有的计算，都是由output操作触发的，比如print。如果没有任何的output操作，那么压根就不会执行定义的计算逻辑。
此外，即使你使用了foreachRDD output操作，也必须在里面对RDD执行action操作，触发对每一个batch的计算逻辑。否则，光有FreachRDD output操作，在里面没有对RDD执行action操作，也不会触发任何逻辑
2.foreachRDD详解
通常在foreachRDD中，都会创建一个connection，比如JDBC Connection ,然后通过connection将数据写入外部存储
误区一：在RDD的foreach操作外部，创建connection
这种方式是错误的，因为他会导致connection对象被序列化后传输到每一个task中。而这种connection对象，实际上一般是不支持序列化的，也就无法被传输
dstream.foreachRDD{Rdd =>
val connection =createNewConnection()
Rdd.foreach{record =>connection.send(record)}
}
误区二：在RDD的foreach操作内部，创建connection
这种方式是可以的，但是效率低下。他会导致对于一个RDD中的每一条数据，都会创建一个connection对象。而通常来说，connection的创建，是很消耗性能的
Dstream.foreachRDD{ rdd =>
Rdd.foreach{record =>
Val connection = createNewConnection()
Connection.send(record)
Connection.close()
}
}
合理的方式一：使用RDD的foreachPartition操作，并且在该操作的内部，创建connection对象，这就相当于是，为RDD的每个partition创建一个connection对象节省资源的多了。
dstream.foreachRDD{rdd =>
rdd.foreachPartition{partitionOfRecords =>
Val connection = createNewConnection()
partitionOfRecores.foreach(recore =>connection.send(record))
Connection.close()
}
}
合理的方式二：自己手动封装一个静态连接池，使用RDD的foreachPartition操作，并且在该操作内部，从静态连接池中，通过静态的方法，获取一个连接，使用之后，在还回去，这样的话，甚至在多个RDD的partition之间，也可以复用连接了。而且可以让连接池采取懒创建的策略，并且空闲一段时间后，将其释放掉
dstream.foreachRDD{rdd =>
rdd.foreachPartition{partitionOfRecoreds =>
Val connection =ConnectionPool.getConnection()
partitionIfRecores.foreach(record => connection.send(record))
ConnectionPool.returnConnection(connection)
}
}
案例：
改写updatestatebykeyworldcount,将每次统计出来的全局的单词计数，写入一份到MySQL数据库中
数据库中建表
Create table wordcount(
Id integer auto_increment primary key,
Updated_time timestamp NOT NULL default CURRENT_TIMESTAMP on update CURRENT_TIMESTAMP,
Word varchar(255),
Count integer
);
九、与spark sql结合
Spark streaming最强大的地方在与可以与spark core、spark sql整合使用，之前通过transform、foreachRDD等算子看到，如何将DStream中的RDD使用spark core执行批处理操作，现在就来看看，如何将DStream中的RDD与spark sql结合后起来使用
案例：每隔10秒。统计最近60秒的，每个种类的每个商品的点击次数，然后统计出每个种类的top3热门商品
|
package com.spark.spark_streaming; import java.util.ArrayList; import java.util.List; import org.apache.spark.SparkConf; import org.apache.spark.api.java.JavaPairRDD; import org.apache.spark.api.java.JavaRDD; import org.apache.spark.api.java.function.Function; import org.apache.spark.api.java.function.Function2; import org.apache.spark.api.java.function.PairFunction; import org.apache.spark.sql.DataFrame; import org.apache.spark.sql.Row; import org.apache.spark.sql.RowFactory; import org.apache.spark.sql.hive.HiveContext; import org.apache.spark.sql.types.DataTypes; import org.apache.spark.sql.types.StructField; import org.apache.spark.sql.types.StructType; import org.apache.spark.streaming.Durations; import org.apache.spark.streaming.api.java.JavaPairDStream; import org.apache.spark.streaming.api.java.JavaReceiverInputDStream; import org.apache.spark.streaming.api.java.JavaStreamingContext; import scala.Tuple2; /** * 与spark sql整合使用，top3热门实时统计 * * @author Administrator * */ public class Top3HotProduct { @SuppressWarnings("deprecation") public static void main(String[] args) { SparkConf conf = new SparkConf().setMaster("local[2]").setAppName("Top3HotProduct"); JavaStreamingContext jsc = new JavaStreamingContext(conf, Durations.seconds(5)); // 首先看一下日志的格式 // username pruduct category // 其实socket是为了方便测试，企业中使用的是kafka // 获取输入数据流 JavaReceiverInputDStream<String> productClickLogsDStream = jsc.socketTextStream("localhost", 9999); // 做一个映射(category_pruduct,1)的格式，可以使用window操作，进行reduceByKey操作，从而统计出每个种类商品的点击次数 JavaPairDStream<String, Integer> categoryPruductPairsDStream = productClickLogsDStream .mapToPair(new PairFunction<String, String, Integer>() { private static final long serialVersionUID = 1L; @Override public Tuple2<String, Integer> call(String productClickLog) throws Exception { return new Tuple2<String, Integer>( productClickLog.split(" ")[2] + "_" + productClickLog.split(" ")[1], 1); } }); // 执行window操作 // 到这里，就可以做到，每隔10秒。对最近60秒的数据，执行reduceByKey操作 // 计算出来这60秒内，每个种类的每个商品的点击次数 JavaPairDStream<String, Integer> categoryProductCountDStream = categoryPruductPairsDStream .reduceByKeyAndWindow(new Function2<Integer, Integer, Integer>() { private static final long serialVersionUID = 1L; @Override public Integer call(Integer v1, Integer v2) throws Exception { return v1 + v2; } }, Durations.seconds(60), Durations.seconds(10)); // 然后真对每个种类的每个商品的点击次数，foreachRDD，在内部使用sparksql执行top3热门商品的统计 categoryProductCountDStream.foreachRDD(new Function<JavaPairRDD<String, Integer>, Void>() { private static final long serialVersionUID = 1L; @Override public Void call(JavaPairRDD<String, Integer> categoryProductCountsRDD) throws Exception { // 将该RDD，转换为javaRDD<Row>的格式 JavaRDD<Row> categoryProductCountRowRDD = categoryProductCountsRDD .map(new Function<Tuple2<String, Integer>, Row>() { private static final long serialVersionUID = 1L; @Override public Row call(Tuple2<String, Integer> categoryProductCount) throws Exception { String category = categoryProductCount._1.split("_")[0]; String product = categoryProductCount._1.split("_")[1]; Integer count = categoryProductCount._2; return RowFactory.create(category, product, count); } }); // 然后执行dataframe转换 List<StructField> structFields = new ArrayList<StructField>(); structFields.add(DataTypes.createStructField("category", DataTypes.StringType, true)); structFields.add(DataTypes.createStructField("product", DataTypes.StringType, true)); structFields.add(DataTypes.createStructField("click_count", DataTypes.IntegerType, true)); StructType structType = DataTypes.createStructType(structFields); HiveContext sqlContext = new HiveContext(categoryProductCountsRDD.context()); DataFrame categoryProductCountDF = sqlContext.createDataFrame(categoryProductCountRowRDD, structType); // 将60秒内的每个种类的，每个商品的点击次数注册为一张零时表 categoryProductCountDF.registerTempTable("product_clock_log"); // 执行sql语句，真对临时表，统计出来每个种类下，点击次数排名前3 的热门商品 DataFrame top3ProductDF = sqlContext.sql("select category,product,click_count from " + "( select category,product,click_count,row_number() over (partition by category order by click_count desc) rank from product_clock_log" + ") tmp where rank <=3"); // 其实在企业中不是打印，应该是保存到redis缓存中，或者是MySQL中 // 然后配合J2EE系统，进行数据的展示和查询、图形报表 top3ProductDF.show(); return null; } }); jsc.start(); jsc.awaitTermination(); jsc.close(); } } |
Scala版本：
|
package com.spark.spark_streaming import org.apache.spark.SparkConf import org.apache.spark.streaming.StreamingContext import org.apache.spark.streaming.Seconds import org.apache.spark.sql.Row import org.apache.spark.sql.types.StructType import org.apache.spark.sql.types.StructField import org.apache.spark.sql.types.StringType import org.apache.spark.sql.types.IntegerType import org.apache.spark.sql.hive.HiveContext object Top3HotProduct { def main(args: Array[String]): Unit = { val conf = new SparkConf().setMaster("local[2]").setAppName("Top3HotProduct") val ssc = new StreamingContext(conf,Seconds(5)) val productClickLogsDStream =ssc.socketTextStream("localhost", 9999) val categoryProductPairsDStream = productClickLogsDStream.map(productLog => (productLog.split(" ")(2)+"_"+productLog.split(" ")(2),1)) val categoryProductCountsDStream = categoryProductPairsDStream. reduceByKeyAndWindow((v1 :Int ,v2 :Int) => v1 + v2, Seconds(60), Seconds(10)) categoryProductCountsDStream.foreachRDD(categoryProductCountsRDD =>{ val categoryProductCountsRowRDD = categoryProductCountsRDD.map(tuple => { val category = tuple._1.split("_")(0) val product =tuple._1.split(" ")(1) val count =tuple._2 Row(category,product,count) })
val structType = StructType(Array( StructField("category",StringType,true), StructField("product",StringType,true), StructField("click_count",IntegerType,true) )) val hiveContext = new HiveContext(categoryProductCountsRDD.context)
val categpryProductCountDF =hiveContext.createDataFrame(categoryProductCountsRowRDD, structType)
categpryProductCountDF.registerTempTable("product_click_log")
val top3ProductDF = hiveContext.sql("select category,product,click_count from " + "( select category,product,click_count,row_number() over (partition by category order by click_count desc) rank from product_clock_log" + ") tmp where rank <=3") top3ProductDF.show()
}) } } |
十、缓存及持久化机制
与RDD 类似，spark streaming也可以进行让开发语言手动控制 ，将数据流中的数据持久化到内存中，对DStream调用persist方法，就可以进行spark streaming自动将数据流中的所产生的RDD，都持久化到内存中，如果要对一个DStream多次进行操作，那么对DStream持久化操作是非常有用的。因为多次操作，可以共享使用内存中的一份缓存数据。
对于基于窗口的操作，比如reduceByKeyAndWindow，以及基于状态的操作，比如updateStateByKey,默认就隐式开启了持久化机制。即，spark streaming默认将上述操作产生的DStream中的数据，缓存到内存中，不需要开发人员手动调用persist方法
对于通过网络接收数据的输入流，比如socket kafka flume等。默认是持久化级别，将数据复制一份，以便于容错。相当于是，用的是类似MEMORY_ONLY_SER_2
与RDD不同，默认的持久化级别，统一都是要序列化的
十一、CheckPoint机制
每一个spark streaming应用程序正常来说，都是要7*24小时（意思就不时时刻刻都是不停的）运转的。 这就是实时计算程序的特点。因为要程序持续不断的对数据进行计算。因此，对实时计算应用的要求，应该是必须要能够与应用程序无关的失败，进行容错
如果要实现这个目标，spark streaming程序就必须将足够的信息checkpoint到容错的存储的存储系统上，从而能够从失败中进行恢复，有两种数据需要进行checkpoint
元数据checkpoint--将定义了流式计算逻辑的信息，保存到容错的存储系统上，比如hdfs，当运行sparkstreaming应用程序的Driver进程所在的节点失败时，该信息可以用于进行恢复，元数据信息包括了：
配置信息：创建sparkstreaming应用程序的配置信息，比如sparkConf中的信息
DStream的操作信息：定义了一个spark stream应用程序中的计算逻辑的DStream操作信息。
未处理的batch信息：那些job正在排队，还没有处理的batch信息
元数据checkpoint--将实时计算过程中产生的RDD数据保存到可靠地存储系统中。
对于一些将多个batch的数据进行聚合的，有状态的transformation操作，这是非常有用的 。在这种transformation操作中，生成的RDD是依赖于batch的RDD的，这会导致随着时间的推移，RDD的依赖链条变得越来越长
要避免由于依赖链条越来越长，导致的一起变的越来越长的失败恢复数据，有状态的transformation操作执行过程中间产生的RDD，会定期的被checkpoint到可靠地存储系统上，比如HDFS。从而削减RDD的依赖链条，进而缩短失败恢复时，RDD的恢复时间
一句话，元数据checkpoint主要是为了从driver失败中进行恢复；而RDDcheckpoint主要是为了使用到有状态的transformation操作时，能够在其生产出的数据丢失时，进行快速的失败恢复
何时使用Checkpoint机制
使用有状态的transformation操作--比如updateStateByKey,或者reduceByKeyAndWindow操作，被使用了，那么checkpoint目录要求是必须提供的，也就是必须开启checkpoint机制，从而进行周期性的RDD checkpoint
要保证可以从Driver失败中进行恢复--元数据checkpoint需要启动，来进行这种情况的恢复
要注意的是，并不是说，所有的spark streaming应用程序，都要启用checkpoint机制，如果即不强制要求从Driver中自动进行恢复，又没有有状态的transformation才做，那么就不需要启动checkpoint。事实上，这么做反而是有助于提升性能的。
如何启用checkpoint机制
1.对于有状态的transformation操作，启用checkpoint机制，定期将其产生的RDD数据checkpoint，是比较简单的
可以通过配置一个容错的、可靠的文件系统中（HDFS）的目录，来启用checkpoint机制，checkpoint数据就会写入该目录。使用streamingContext的checkpoint()方法即可。然后，你就可以放心使用有状态的transformation操作了
如果为了要从Driver失败中进行恢复，那么启用checkpoint机制比较复杂的。需要改写spark streaming应用程序
当应用程序第一次启动的时候，需要创建一个新的streamingContext，并且调用start()方法，进行启动。当Driver从失败中恢复过来的时候，需要从checkpoint目录中记录的元数据中，恢复出来一个sparkContext
为driver失败的恢复机制重写程序（java）
|
JavaStreamingContextFactory jscf = new JavaStreamingContextFactory (){ public JavaStreamingContext create(){ JavaStreamingContext jsc = new JavaStreamingContext (..); JavaDSream<String> lines =jscf.socketTextStream(...); Jscf.chechpoint(checkPointDirectory); Return jscf; } }; JavaStreamingContext context = JavaStreamingContext.getOrCreate(checkpointDirectory,contextFatory); Context.start(); context.awaitTermina(); |
Scala版本：
|
def functionToCtreateContext():StreamingContext ={ Val ssc = new StreamingContext(...) Val lines = ssc.socketTextStream(...) Ssc.checkpoint(checkpointDirectory) ssc } Val context = StreamingContext.getOrCreate(checkpointDirectory,functionToCtreateContext_) Conntext.start() context.awaitTermination() |
配置spark-submit提交参数
按照上面的方法，进行spark streaming应用程序的重写后。当第一次 运行程序的时候，如果发现checkpoint不存在，那么就使用定义的函数第一次创建一个StreamingContext，并将其元数据写入到checkpoint目录；
当从driver失败中恢复过来的时候，发现checkpoint目录已经在了，那么会使用该目录中的元数据创建一个streamingContext.
但是上面的重写应用程序的过程，只是实现driver失败自动恢复的第一步，第二步是，必须确保driver可以在失败时，自动被重启。
能够自动的从driver失败中恢复过来，运行sparkstreaming应用程序的集群，就必须监控driver运行的过程。 并且在他失败时将他重启，对于spark 自身的standalone模式，需要进行一些配货站去supervise driver，在他失败时将其重启
首先，要在spark-submit中，添加--deploy-mode参数，默认将其值为client，即在提交应用的机器上启动driver，但是，要能够自动重启driver，就必须将其值设置为cluster，此外，需要添加--supervise参数
使用上述第二步骤提交应用程序后，就可以让driver在失败的时候被重启，并且通过checkpoint目录中的元数据恢复streamingContext
Checkpoint的说明;
将RDDcheckpoint到可靠的存储系统行，会耗费很多的性能。当RDD被checkpoint时，对导致这些batch的处理时间增加。因此，checkpoint的间隔，需要谨慎的设置。对于那些间隔很多的batch，如果一秒，如果还要执行checkpoint操作，则会大幅度的削减吞吐量，而另一方面，如果checkpoint的操作执行的不太频繁，就会导致RDD的lineage变长，又会有失败恢复时间过长的风险
对于那些时间要求checkpoint的有状态的transformation操作，默认的checkpoint间隔通常是batch的整数倍，至少是10秒。使用功能DStream的checkpoint方法，可以设置这个DStream的checkpoint的间隔时长，通常来说，将checkpoint间隔设置为窗口操作的滑动间隔的5~10倍，是个不错的选择。
十二、部署、升级和监控应用程序
1.部署应用程序
1.有一个资源管理器，比如standalone模式下的spark集群，yarn模式下的yarn集群等
2.打包应用程序为一个jar包
3.为executor配置充足的内存，因为receiver接受到的数据，是要存储在executor的内存中的，所以executor必须配置足够的内存来保存接收到的数据。要注意的是，如果你要执行的窗口的长度为10分钟的窗口操作，那么executor的内存资源就必须足够保存10分钟以内的数据，因此内存的资源要求是取决于你执行的操作的
4.配置checkpoint，如果你的应用程序要求checkpoint操作，那么就必须配置一个hadoop兼容的文件系统，比如hdfs的目录作为checkpoint目录
5.配置driver的自动恢复，没如果要让driver在失败时，自动恢复，之前已经讲过，一方面，要重写driver程序，一方面，要在spark-submit中添加参数
启用预写日志：
简写为WAL，全称为write ahead Log.从spark1.2版本开始，就引入了基于容错机制的文件系统的WAL机制。如果启用该机制，receiver接收到的所有的数据都会被写入配置的checkpoint目录中的预写日志。这种机制可以让driver在恢复的时候，避免数据的丢失，并且可以确保整个实时计算的过程中，零数据丢失
要配置该机制，首先要调用streamingContext的checkpoint方法，设置一个checkpoint目录。然后需要将spark.streaming.receiver.writeAheadLog.eanable参数设置为true
然而，这种极强的可靠性机制，会导致receiver的吞吐量大幅度的下降，因为单位时间呗，有相当一部分时间需要将数据写入预写日志。如果又希望开启预写日志，确保数据零丢失，又不希望影响系统的吞吐量，那么可以创建多个输入dstream，启动多个receiver
此外，在启用了预写日志之后，推荐将复制持久化机制禁用掉。因为所有的数据已经保存在容错的文件系统中了，不需要在用复制机制进行持久化，保存一份副本了。只要将输入DStream的持久化机制设置一下即可，persist(StoragLevel.MEMORY_AND_DISK_SER).（默认是基于复制的持久化策略）
设置receiver接收速度
如果集群资源有限，并没有达到足以让应用程序一接收到数据，就理解处理他，Receiver可以被设置一个最大接收限速，以每秒接收多少条单位来限制
spark.streaming.receiver.maxRate和spark.streaming.kafka.maxRatePerPartition参数可以用来设置，前者设置普通Receiver，后者kafka Direct
Spark1.5中，对于kakfa Direct方式，引入了backpressure机制，从而不需要设置Receiver的限制，spark 可以自动估计receiver最合理的接收速度，并根据情况动态调整，只要将spark.streaming.backpressure.enable设置为true即可。
在企业实际应用场景中，通常是推荐kafka Direct方式的，特别是现在随着spark版本的提升，越来越完善这个kafka Direct机制。
优点：
1.不用receiver，不会独占集群的一个cpu core
2.有backpressure自动调节接收速率的机制
升级应用程序
由于spark streaming应用程序都是7*24小时运行的。因此如果需要对正在运行的应用程序，进行代码的升级，那么有两种方式可以实现
1.升级后的spark应用程序直接启动，先与旧的spark应用程序并行执行。当确保新的应用程序启动没有问题之后，就可以将旧的应用程序给停掉。但是要注意的是，这种方式只适用于，能够允许多个客户端读取各自独立的数据，也就是读取相同的数据
2.小心的关闭已经正在运行的应用程序，使用streamingContext的stop方法，可以确保接收到的数据都处理完之后，才停止。然后将升级后的程序部署上去，启动。这样就可以确保中间的数据丢失和未处理。因为新的应用程序会从老的应用程序为消费到的地方，继续消费。
但是注意，这种方式必须是支持数据缓存的数据源才可以，比如kafka、flume等。如果数据源不支持数据缓存，那么会导致数据丢失。
注意：配置driver自动恢复机制是，如果想要根据旧的应用程序的checkpoint信息，启动新的应用程序，是不可行的。需要让新的应用程序针对新的checkpoint目录启动，或者删除之前的checkpoint目录。
监控应用程序
当spark steaming应用启动的时候，spark web ui会显示一个独立的streaming tab，会显示receiver的信息，比如是否活跃，接收到了的数据，是否有异常等，还会显示完成的batch的信息，batch的处理时间，队列延迟等。这些信息可以用于检测spark steaming应用的进度
Spark UI中，以下的两个统计指标格外的重要
1,处理时间--每个batch的数据的处理耗时
2,调度延迟--一个batch在队列中阻塞住，等待上一个batch完成处理的时间
如果batch的处理时间，比batch的间隔要长的话，而且调度延迟时间持续增长，应用程序不足以使用当前的设定的速度来处理接收到的数据，此时，可以考虑增加batch的间隔时间。
容错机制以及事务语义详解
容错机制背景
Spark RDD的基础的容错语义：
1.RDD，ressilient Distributed Dataset，是不可变的、确定的、可以重新计算的、分布式的数据集。每一个RDD都会记住确定好的计算操作的血缘关系，这些操作应用在一个容错
的数据集来创建RDD
2.如果因为某个worker节点的失败，导致RDD的某个partition数据丢失，那么那个partition可以通过对原始数据Juin应用操作血缘关系，来重新计算
3.所有的RDD transformation操作是确定的，最后一个被转换出来的RDD的数据，一定是不会因为spark集群的失败而丢失的
Spark操作的通常是容错文件系统中的数据，如hdfs。因此，所有的容错数据生成的RDD也是容错的。然而对于spark streaming来说，这却行不通，因为在大多数的情况下，数据都是通过网络接收到的（除了使用filestream）。要让spark streaming程序中，所有生成的RDD，都达到与普通spark程序的RDD相同的容错性，就收到的数据必须被复制到多个worker节点上的executor内存中，默认的复制因子是2。
基于上述的理论，在出现失败的事件时，有两种数据需要被恢复：
1.数据收集到了，并且已经被复制过了--这种数据在一个worker节点挂掉之后，是可以继续存活的，因为在其他的worker节点上，还有他的一个副本。
2.数据接收到了，但是正在缓存中，等待复制的--因为还没有复制该数据，因此恢复他的唯一的方式就是重新从数据源获取一份
此外，还有两种失败使我们需要进行考虑的：
1.worker节点的失败
任何一个运行了executor的worker节点的挂掉，都会导致该节点上所有的内存中的数据都丢失，如果有receiver运行在该worker节点上的executor中，那么缓存的，带复制的数据，都会丢失
2.Driver节点的失败
如果运行spark streaming应用程序的driver节点失败了，那么显然sparkcontext会丢失，那么该application的所有的executor数据都会丢失。
Sparkstreaming语义：
流式计算系统的容错语义，通常是以一条记录能够被处理多少次来衡量。有三种可以提供：
1最多一次.：每条数据，可能会被处理一次或者不会被处理。可能有数据丢失
2.至少一次：每条数据被处理一次或者多次，这种语义比最多一次要更强大，因为他确保数据零丢失，但是可能导致记录被重置处理几次
3.一次且仅一次：每条记录只会被处理一次---没有数据丢失，并且没有数据会被处理多次。这是最强的一种容错语义
Sparkstreaming的基础容错语义：
1.接收数据，使用receiver或其他方式接收数据
2.计算数据，使用dstream的transformation操作对数据进行计算和处理
3.推送数据，最后计算的数据会被推送到外部系统中，如文件系统、数据库等等
如果应用程序要求必须有一次且仅仅一次的语义，那么将上述的三个步骤都必须提供一次且仅一次的语义。每条数据都得保证，只能接受一次，只能计算一次、只能推送一次。
Spark streaming中执行这些语义的步骤如下：
1.接受数据：不同的数据源提供不同的语义保障
2.计算数据：所有接收到的数据一定只会被执行一次，这是基于RDD的基础语义锁保障的。即使有失败，只要接收到的数据还是可访问的，最后一个计算出来的数据一定是相同的
3.推送数据：output操作默认能确保至少一次的语义，因为它依赖output操作的类型，以及底层系统的语义支持（比如是否有事务的支持等），但是用户可以实现他们自己的事务机制。来确保一次且仅一次的语义
接收数据的语义：
1.基于文件的数据源
如果所有的输入源数据都在一个容错的文件系统中，如hdfs、spark streaming一定可以从失败进行恢复，并且处理所有的数据。这就提供了一次且仅一次的语义，意味着对所有的数据只会处理一次。
2.基于receiver的数据源
对于基于Receiver的数据源，容错语义依赖于失败的场景和receiver类型
可靠的receiver：这种receiver会在接收到了数据，并且将数据复制之后，对数据源执行确认操作，如果receiver在数据接收和复制完成之前，就失败了，那么数据源对于缓存的数据会接收不到确认，此时，当receiver重启之后，数据源会重新发送数据。没有数据会丢失
不可靠的receiver：这种receiver不会发送确认操作，因此worker或者driver节点失败的时候，可能会导致数据丢失。
不同的receiver，提供了不同的语义，如果worker节点失败了，那么使用的是可靠receiver的话，没有数据会丢失。使用的是不可靠的receiver的话，接收到，但是还没有复制的数据，可能会丢失，如果driver节点失败的话，所有过去接收到的和复制过缓存在内存中的数据，全部会丢失
要避免这种过去接收的所有的数据都丢失的问题，spark1.2版本开始，引入了预写日志机制，可以将receiver接收到的数据保存到容错存储中。如果使用可靠的receiver，并且还开启了预写日志的机制，那么可以保证数据的零丢失。这种情况下，会提供至少一次的保障
|
部署场景 |
Worker失效 |
Driver失效 |
|
Spark1.0，没有WAL |
1.可靠receiver，可以保证数据零丢失 2.不可靠receiver，可以保证零丢失数据 3.至少一次的语义 |
1.不可靠receiver，缓存的数据全部丢失 2.任何receiver，过去接收的所有的数据全部丢失 3.没有容错语义 |
|
Spark1.2，开启了WAL |
1.可靠receiver，零数据丢失 2.至少一次的语义 |
1.可靠receiver和文件，零数据丢失 |
从spark1.3版本开始，引入了新的kafka Direct API。可以保证，所有从kafka接收到的数据，都是一次仅一次，基于该语义保障，如果自己在实现一次且仅一次语义的output操作，那么就可以获的整个spark steaming应用程序的一次仅一次的语义
输出数据的容错语义：
Output操作，比如foreachRDD可以提供至少一次的语义。那就意味着，当worker节点失效的时候，转换后的数据可能被写入外部系统一次或多次。对于写入文件来说，这还是可以接受的，因为会覆盖数据。但是真正获得一次且仅一次的语义，有两个方法;
1.幂等更新：多次操作，都是写的是相同的数据
2.事务更新：所有的操作都应该作为事务的，从而让写入执行一次且一次。给每个batch的数据都赋予一个唯一的标识，然后跟新判断的时候。如果数据库中号没有哦唯一标识
Storm的容错语义：
十三、架构原理
十四、Spark streaming性能调优
十五、数据接收原理
十六、数据处理原理
十七、性能调优