# if open wal
org.apache.spark.SparkException: Could not read data from write ahead log record FileBasedWriteAheadLogSegment
SparkStreaming开启了checkpoint wal后有时会出现如上报错，但不会影响整体程序，只会丢失报错的那个job的数据。其根本原因是wal文件被删了，被sparkstreaming自己的清除机制删掉了。通常意味着一定程度流式程序上存在速率不匹配或堆积问题。
查看driver日志可发现类似如下的日志：
-- :: INFO [Logging.scala:] Attempting to clear old log files in hdfs://alps-cluster/tmp/banyan/checkpoint/RhinoWechatConsumer/receivedBlockMetadata older than 1490248380000:
-- :: INFO [Logging.scala:] Attempting to clear old log files in hdfs://alps-cluster/tmp/banyan/checkpoint/RhinoWechatConsumer/receivedBlockMetadata older than 1490248470000: hdfs://alps-cluster/tmp/banyan/checkpoint/RhinoWechatConsumer/receivedBlockMetadata/log-1490248404471-1490248464471
-- :: INFO [Logging.scala:] Cleared log files in hdfs://alps-cluster/tmp/banyan/checkpoint/RhinoWechatConsumer/receivedBlockMetadata older than 1490248470000
-- :: ERROR [Logging.scala:] Task in stage 35.0 failed times; aborting job
-- :: ERROR [Logging.scala:] Error running job streaming job ms.
org.apache.spark.SparkException: Job aborted due to stage failure: Task in stage 35.0 failed times, most recent failure: Lost task 41.3 in stage 35.0 (TID , alps60): org.apache.spark.SparkException: Could not read data from write ahead log record FileBasedWriteAheadLogSegment(hdfs://alps-cluster/tmp/banyan/checkpoint/RhinoWechatConsumer/receivedData/0/log-1490248403649-1490248463649,44333482,118014)
at org.apache.spark.streaming.rdd.WriteAheadLogBackedBlockRDD.org$apache$spark$streaming$rdd$WriteAheadLogBackedBlockRDD$$getBlockFromWriteAheadLog$(WriteAheadLogBackedBlockRDD.scala:)
可以发现 1490248403649 的日志被删除程序删除了（cleared log older than 1490248470000），然后这个wal就报错了。
Spark官方文档没有任何关于这个的配置，因此直接看源码。（spark很多这样的坑，得看源码才知道如何hack或有些隐藏配置）。
1.FileBasedWriteAheadLogSegment 类中根据日志搜索发现了clean方法（后面的逻辑就是具体删除逻辑，暂不关心），核心就是如何调整这个threshTime了。
/**
* Delete the log files that are older than the threshold time.
*
* Its important to note that the threshold time is based on the time stamps used in the log
* files, which is usually based on the local system time. So if there is coordination necessary
* between the node calculating the threshTime (say, driver node), and the local system time
* (say, worker node), the caller has to take account of possible time skew.
*
* If waitForCompletion is set to true, this method will return only after old logs have been
* deleted. This should be set to true only for testing. Else the files will be deleted
* asynchronously.
*/
def clean(threshTime: Long, waitForCompletion: Boolean): Unit = {
val oldLogFiles = synchronized {
val expiredLogs = pastLogs.filter { _.endTime < threshTime }
pastLogs --= expiredLogs
expiredLogs
}
logInfo(s"Attempting to clear ${oldLogFiles.size} old log files in $logDirectory " +
s"older than $threshTime: ${oldLogFiles.map { _.path }.mkString("\n")}")
2.一步步看调用追踪出去，ReceivedBlockHandler -> ReceiverSupervisorImpl -> CleanUpOldBlocks 。这里有个和ReceiverTracker通信的rpc，因此直接搜索CleanUpOldBlocks -> ReceiverTracker -> JobGenerator
在JobGenerator.clearCheckpointData 中有这么一段逻辑
/** Clear DStream checkpoint data for the given `time`. */
private def clearCheckpointData(time: Time) {
ssc.graph.clearCheckpointData(time) // All the checkpoint information about which batches have been processed, etc have
// been saved to checkpoints, so its safe to delete block metadata and data WAL files
val maxRememberDuration = graph.getMaxInputStreamRememberDuration()
jobScheduler.receiverTracker.cleanupOldBlocksAndBatches(time - maxRememberDuration)
jobScheduler.inputInfoTracker.cleanup(time - maxRememberDuration)
markBatchFullyProcessed(time)
}
发现了 ssc.graph有个 maxRememberDuration 的成员属性！这就意味着有机会通过ssc去修改它。
搜索一下代码便发现了相关方法：
jssc.remember(new Duration(2 * 3600 * 1000));
反思：
从之前的日志我们发现默认的清除间隔是几十秒左右，但是在代码中我们可以发现这个参数只能被设置一次（每次设置都会检查当前为null才生效，初始值为null）。所以问题来了，这几十秒在哪里设置的？代码一时没找到，于是项目直接搜索 remember，发现了在DStream里的初始化代码（其中slideDuration初始化来自InputDStream）。根据计算，我们的batchInterval为15s，其他两个没有设置，则checkpointDuration 为15s，rememberDuration为30s。
override def slideDuration: Duration = {
if (ssc == null) throw new Exception("ssc is null")
if (ssc.graph.batchDuration == null) throw new Exception("batchDuration is null")
ssc.graph.batchDuration
}
/**
* Initialize the DStream by setting the "zero" time, based on which
* the validity of future times is calculated. This method also recursively initializes
* its parent DStreams.
*/
private[streaming] def initialize(time: Time) {
if (zeroTime != null && zeroTime != time) {
throw new SparkException("ZeroTime is already initialized to " + zeroTime
+ ", cannot initialize it again to " + time)
}
zeroTime = time
// Set the checkpoint interval to be slideDuration or 10 seconds, which ever is larger
if (mustCheckpoint && checkpointDuration == null) {
checkpointDuration = slideDuration * math.ceil(Seconds(10) / slideDuration).toInt
logInfo("Checkpoint interval automatically set to " + checkpointDuration)
}
// Set the minimum value of the rememberDuration if not already set
var minRememberDuration = slideDuration
if (checkpointDuration != null && minRememberDuration <= checkpointDuration) {
// times 2 just to be sure that the latest checkpoint is not forgotten (#paranoia)
minRememberDuration = checkpointDuration * 2
}
if (rememberDuration == null || rememberDuration < minRememberDuration) {
rememberDuration = minRememberDuration
}
// Initialize the dependencies
dependencies.foreach(_.initialize(zeroTime))
}