大数据-SparkStreaming（六）

                               大数据-SparkStreaming（六）

数据丢失如何处理

利用WAL把数据写入到HDFS中

步骤一：设置checkpoint目录

streamingContext.setCheckpoint(hdfsDirectory)

步骤二：开启WAL日志

sparkConf.set("spark.streaming.receiver.writeAheadLog.enable", "true")

步骤三：需要reliable receiver

当数据写完了WAL后，才告诉数据源数据已经消费，对于没有告诉数据源的数据，可以从数据源中重新消费数据。

步骤四：取消备份

使用StorageLevel.MEMORY_AND_DISK_SER来存储数据源，不需要后缀为2的策略了，因为HDFS已经是多副本了。

Reliable Receiver   : 当数据接收到，并且已经备份存储后，再发送回执给数据源
Unreliable Receiver : 不发送回执给数据源

• WAL

WAL使用在文件系统和数据库中用于数据操作的持久性，先把数据写到一个持久化的日志中，然后对数据做操作，如果操作过程中系统挂了，恢复的时候可以重新读取日志文件再次进行操作。

对于像kafka和flume这些使用接收器来接收数据的数据源。接收器作为一个长时间的任务运行在executor中，负责从数据源接收数据，如果数据源支持的话，向数据源确认接收到数据，然后把数据存储在executor的内存中，然后在exector上运行任务处理这些数据。

如果wal启用了，所有接收到的数据会保存到一个日志文件中去（HDFS), 这样保存接收数据的持久性，此外，如果只有在数据写入到log中之后接收器才向数据源确认，这样driver重启后那些保存在内存中但是没有写入到log中的数据将会重新发送，这两点保证的数据的无丢失。

当一个task很慢容错

开启推测机制：

spark.speculation=true，每隔一段时间来检查有哪些正在运行的task需要重新调度（spark.speculation.interval=100ms），假设总的task有10个，成功的task的数量 > 0.75 * 10（spark.speculation.quantile=0.75），正在运行的task的运行时间 > 1.5 * 成功运行task的平均时间（spark.speculation.multiplier=1.5），则这个正在运行的task需要重新等待调度。

注意：

在分布式环境中导致某个Task执行缓慢的情况有很多，负载不均、程序bug、资源不均、数据倾斜等，而且这些情况在分布式任务计算环境中是常态。Speculative Task这种以空间换时间的思路对计算资源是种压榨，同时如果Speculative Task本身也变成了Slow Task会导致情况进一步恶化。