【Spark篇】---SparkStreaming+Kafka的两种模式receiver模式和Direct模式

一、前述

SparkStreamin是流式问题的解决的代表，一般结合kafka使用，所以本文着重讲解sparkStreaming+kafka两种模式。

二、具体

1、Receiver模式

   原理图：

 receiver模式理解：

在SparkStreaming程序运行起来后，Executor中会有receiver tasks接收kafka推送过来的数据。数据会被持久化，默认级别为MEMORY_AND_DISK_SER_2,这个级别也可以修改。receiver task对接收过来的数据进行存储和备份，这个过程会有节点之间的数据传输。备份完成后去zookeeper中更新消费偏移量，然后向Driver中的receiver tracker汇报数据的位置。最后Driver根据数据本地化将task分发到不同节点上执行。

receiver模式中存在的问题：

当Driver进程挂掉后，Driver下的Executor都会被杀掉，当更新完zookeeper消费偏移量的时候，Driver如果挂掉了，就会存在找不到数据的问题，相当于丢失数据。

如何解决这个问题？

开启WAL(write ahead log)预写日志机制,在接受过来数据备份到其他节点的时候，同时备份到HDFS上一份（我们需要将接收来的数据的持久化级别降级到MEMORY_AND_DISK），这样就能保证数据的安全性。不过，因为写HDFS比较消耗性能，要在备份完数据之后才能进行更新zookeeper以及汇报位置等，这样会增加job的执行时间，这样对于任务的执行提高了延迟度。

<!-- li { list-style: none; margin: 0; } p { margin: 0; } span.l { color: red; font-weight: bold; } a.mapnode:link {text-decoration: none; color: black; } a.mapnode:visited {text-decoration: none; color: black; } a.mapnode:active {text-decoration: none; color: black; } a.mapnode:hover {text-decoration: none; color: black; background: #eeeee0; } -->

注意：1）开启WAL之后，接受数据级别要降级，有效率问题。2）开启WAL要checkpoint 3）开启WAL(write ahead log),往HDFS中备份一份数据

<!-- li { list-style: none; margin: 0; } p { margin: 0; } span.l { color: red; font-weight: bold; } a.mapnode:link {text-decoration: none; color: black; } a.mapnode:visited {text-decoration: none; color: black; } a.mapnode:active {text-decoration: none; color: black; } a.mapnode:hover {text-decoration: none; color: black; background: #eeeee0; } -->

<!-- li { list-style: none; margin: 0; } p { margin: 0; } span.l { color: red; font-weight: bold; } a.mapnode:link {text-decoration: none; color: black; } a.mapnode:visited {text-decoration: none; color: black; } a.mapnode:active {text-decoration: none; color: black; } a.mapnode:hover {text-decoration: none; color: black; background: #eeeee0; } -- -->

receiver的并行度设置

receiver的并行度是由spark.streaming.blockInterval来决定的，默认为200ms,假设batchInterval为5s,那么每隔blockInterval就会产生一个block,这里就对应每批次产生RDD的partition,这样5秒产生的这个Dstream中的这个RDD的partition为25个，并行度就是25。如果想提高并行度可以减少blockInterval的数值，但是最好不要低于50ms。