org.apache.spark.streaming.util

Class WriteAheadLogManager

Object

org.apache.spark.streaming.util.WriteAheadLogManager

All Implemented Interfaces:

Logging

public class WriteAheadLogManager
extends Object
implements Logging

This class manages write ahead log files. - Writes records (bytebuffers) to periodically rotating log files. - Recovers the log files and the reads the recovered records upon failures. - Cleans up old log files.

Uses WriteAheadLogWriter to write and WriteAheadLogReader to read.

Nested Class Summary

Nested Classes

Modifier and Type	Class and Description
static class	WriteAheadLogManager.LogInfo
static class	WriteAheadLogManager.LogInfo$

Constructor Summary

Constructors

Constructor and Description
WriteAheadLogManager(String logDirectory, org.apache.hadoop.conf.Configuration hadoopConf, int rollingIntervalSecs, int maxFailures, String callerName, Clock clock)

Method Summary

Methods

Modifier and Type	Method and Description
void	cleanupOldLogs(long threshTime, boolean waitForCompletion) Delete the log files that are older than the threshold time.
static scala.util.matching.Regex	logFileRegex()
static scala.collection.Seq<WriteAheadLogManager.LogInfo>	logFilesTologInfo(scala.collection.Seq<org.apache.hadoop.fs.Path> files) Convert a sequence of files to a sequence of sorted LogInfo objects
scala.collection.Iterator<java.nio.ByteBuffer>	readFromLog() Read all the existing logs from the log directory.
void	stop() Stop the manager, close any open log writer
static String	timeToLogFile(long startTime, long stopTime)
WriteAheadLogFileSegment	writeToLog(java.nio.ByteBuffer byteBuffer) Write a byte buffer to the log file.

Methods inherited from class Object

equals, getClass, hashCode, notify, notifyAll, toString, wait, wait, wait

Methods inherited from interface org.apache.spark.Logging

initializeIfNecessary, initializeLogging, isTraceEnabled, log_, log, logDebug, logDebug, logError, logError, logInfo, logInfo, logTrace, logTrace, logWarning, logWarning

Constructor Detail

WriteAheadLogManager

public WriteAheadLogManager(String logDirectory,
                    org.apache.hadoop.conf.Configuration hadoopConf,
                    int rollingIntervalSecs,
                    int maxFailures,
                    String callerName,
                    Clock clock)

Method Detail

logFileRegex

public static scala.util.matching.Regex logFileRegex()

timeToLogFile

public static String timeToLogFile(long startTime,
                   long stopTime)

logFilesTologInfo

public static scala.collection.Seq<WriteAheadLogManager.LogInfo> logFilesTologInfo(scala.collection.Seq<org.apache.hadoop.fs.Path> files)

Convert a sequence of files to a sequence of sorted LogInfo objects

writeToLog

public WriteAheadLogFileSegment writeToLog(java.nio.ByteBuffer byteBuffer)

Write a byte buffer to the log file. This method synchronously writes the data in the ByteBuffer to HDFS. When this method returns, the data is guaranteed to have been flushed to HDFS, and will be available for readers to read.

readFromLog

public scala.collection.Iterator<java.nio.ByteBuffer> readFromLog()

Read all the existing logs from the log directory.

Note that this is typically called when the caller is initializing and wants to recover past state from the write ahead logs (that is, before making any writes). If this is called after writes have been made using this manager, then it may not return the latest the records. This does not deal with currently active log files, and hence the implementation is kept simple.

cleanupOldLogs

public void cleanupOldLogs(long threshTime,
                  boolean waitForCompletion)

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.

stop

public void stop()

Stop the manager, close any open log writer