六、SparkStreaming窗口操作
Spark提供了一组 窗口操作，通过滑动窗口技术对大规模数据的增量更新进行统计分析
• Window Operation： 定时进行一定时间段内的数据处理(上图time 3 4 5 每个2秒，一共6秒)
• 任何基于窗口操作需要指定两个参数：
– 窗口总长度(window length)10s
– 滑动时间间隔(slide interval) 2s
执行代码前先启动nc -lk 9999
执行代码：整个窗口长度10s,每2秒打印一次
改代码：
1.上代码改为seconds(10),second(3),报错，必须为scc seconds的整数倍
2.改为seconds(9),second(2)也出错，如下图，也必须为上scc seconds整数倍
七、Sparkstreaming全局统计量
• 如果要使用updateStateByKey算子，就必须设置一个checkpoint目录，开启checkpoint机制
• 这样的话才能把每个key对应的state除了在内存中有，那么是不是也要checkpoint一份
• 因为你要长期保存一份key的state的话，那么spark streaming是要求必须用checkpoint的，以便于在内存数据丢失的时候，可以从checkpoint中恢复数据
左10s统计结果，右再过2s统计结果，最后 全局合并统计updatestatebykey,要开启checkpoint且先nc -lp 9999
再输入7个a后
Sparkstreaming容错性分析
(RDD容错靠血缘关系DAG，sparkstreaming靠WAL)
• 实时的流式处理系统必须是7*24运行的，同时可以从各种各样的系统错误中恢复，在设计之初，Spark Streaming就支持driver和worker节点的错误恢复。
1. Worker容错：spark和rdd的保证worker节点的容错性。spark streaming构建在spark之上，所以它的worker节点也是同样的容错机制
•2.Driver容错：依赖WAL(WriteAheadLog)持久化日志
– 启动WAL需要做如下的配置
– 1：给streamingContext设置checkpoint的目录，该目录必须是HADOOP支持的文件系统hdfs，用来保存WAL和
做Streaming的checkpoint
– 2：spark.streaming.receiver.writeAheadLog.enable 设置为true； receiver才有WAL
Sparkstreaming中WAL简介
• Spark应用分布式运行的，如果driver进程挂了，所有的executor进程将不可用，保存在这些进程所
持有内存中的数据将会丢失。为了避免这些数据的丢失，Spark Streaming中引入了一个WAL.
• WAL在文件系统和数据库中用于数据操作的持久化，先把数据写到一个持久化的日志中，然后对数
据做操作，如果操作过程中系统挂了，恢复的时候可以重新读取日志文件再次进行操作。
• 如果WAL 启用了，所有接收到的数据会保存到一个日志文件中去(HDFS), 这样保存接收数据的持
久性，此外，如果只有在数据写入到log中之后接收器才向数据源确认，这样drive重启后那些保存在
内存中但是没有写入到log中的数据将会重新发送，这两点保证的数据的无丢失。
WAL工作原理
driver=AM , executor=worker节点
block+文件数据(代码)WAL 两部分结合
1. 蓝色的箭头表示接收的数据：
– 接收器把数据流打包成块，存储在executor的内存中，如果开启了WAL，将会把数据写入到存在容错文
件系统的日志文件中
2. 青色的箭头表示提醒driver：
– 接收到的数据块的元信息发送给driver中的StreamingContext, 这些元数据包括：executor内存中数据
块的引用ID和日志文件中数据块的偏移信息
3. 红色箭头表示处理数据：
– 每一个批处理间隔，StreamingContext使用块信息用来生成RDD和jobs. SparkContext执行这些job用
于处理executor内存中的数据块
4. 黄色箭头表示checkpoint这些计算：
– 以便于恢复。流式处理会周期的被checkpoint到文件中
Sparkstreaming消费kafka
Spark Streaming 接受数据的方式有两种： 只有receive有wal,direct不需要
• Receiver-based Approach：offset存储在zookeeper，由Receiver维护，Spark获取数据存入executor中，调用
Kafka高阶API
• Direct Approach (No Receivers)：offset自己存储和维护，由Spark维护，且可以从每个分区读取数据，调用Kafka低阶API
SparkstreamingonKafkaDirect
1. Direct的方式是会直接操作kafka底层的元数据信息
2. 由于直接操作的是kafka，kafka就相当于底层的文件系统(对应receiver的executor内存)。
3. 由于底层是直接读数据，没有所谓的Receiver，直接是周期性(Batch Intervel)的查询kafka，
处理数据的时候，我们会使用基于kafka原生的Consumer api来获取kafka中特定范围(offset
范围)中的数据。
4. 读取多个kafka partition，Spark也会创建RDD的partition ，这个时候RDD的partition和
kafka的partition是一致的。
5. 不需要开启wal机制，从数据零丢失的角度来看，极大的提升了效率，还至少能节省一倍的磁盘
空间。从kafka获取数据，比从hdfs获取数据，因为zero copy的方式，速度肯定更快。
Direct与 Receiver对比
• 从容错角度：
– Receiver(高层次的消费者API)：在失败的情况下，有些数据很有可能会被处理不止一次。 接收到的数
据被可靠地保存到WAL中，但是还没有来得及更新Zookeeper中Kafka偏移量。导致数据不一致性：
Streaming知道数据被接收，但Kafka认为数据还没被接收。这样系统恢复正常时，Kafka会再一次发送这
些数据。at least once
– Direct(低层次消费者API)：给出每个batch区间需要读取的偏移量位置，每个batch的Job被运行时，
对应偏移量的数据从Kafka拉取，偏移量信息也被可靠地存储(checkpoint)，在从失败中恢复可以直接
读取这些偏移量信息。exactly once
Direct API消除了需要使用WAL的Receivers的情况，而且确保每个Kafka记录仅被接收一次并被高效地接收
。这就使得我们可以将Spark Streaming和Kafka很好地整合在一起。总体来说，这些特性使得流处理管道拥
有高容错性，高效性，而且很容易地被使用。
