[SPARK-7056][Streaming] Make the Write Ahead Log pluggable #5645
There are no files selected for viewing
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,59 @@ | ||
| /* | ||
| * Licensed to the Apache Software Foundation (ASF) under one or more | ||
| * contributor license agreements. See the NOTICE file distributed with | ||
| * this work for additional information regarding copyright ownership. | ||
| * The ASF licenses this file to You under the Apache License, Version 2.0 | ||
| * (the "License"); you may not use this file except in compliance with | ||
| * the License. You may obtain a copy of the License at | ||
| * | ||
| * http://www.apache.org/licenses/LICENSE-2.0 | ||
| * | ||
| * Unless required by applicable law or agreed to in writing, software | ||
| * distributed under the License is distributed on an "AS IS" BASIS, | ||
| * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. | ||
| * See the License for the specific language governing permissions and | ||
| * limitations under the License. | ||
| */ | ||
|
|
||
| package org.apache.spark.streaming.util; | ||
|
|
||
| import java.nio.ByteBuffer; | ||
| import java.util.Iterator; | ||
|
|
||
| /** | ||
| * This abstract class represents a write ahead log (aka journal) that is used by Spark Streaming | ||
| * to save the received data (by receivers) and associated metadata to a reliable storage, so that | ||
| * they can be recovered after driver failures. See the Spark documentation for more information | ||
| * on how to plug in your own custom implementation of a write ahead log. | ||
| */ | ||
| @org.apache.spark.annotation.DeveloperApi | ||
| public abstract class WriteAheadLog { | ||
| /** | ||
| * Write the record to the log and return the segment information that is necessary to read | ||
| * back the written record. The time is used to the index the record, such that it can be | ||
| * cleaned later. Note that the written data must be durable and readable (using the | ||
| * segment info) by the time this function returns. | ||
| */ | ||
| abstract public WriteAheadLogRecordHandle write(ByteBuffer record, long time); | ||
|
|
||
| /** | ||
| * Read a written record based on the given segment information. | ||
| */ | ||
| abstract public ByteBuffer read(WriteAheadLogRecordHandle handle); | ||
|
|
||
| /** | ||
| * Read and return an iterator of all the records that have been written but not yet cleaned up. | ||
| */ | ||
| abstract public Iterator<ByteBuffer> readAll(); | ||
|
|
||
| /** | ||
| * Clean all the records that are older than the threshold time. It can wait for | ||
| * the completion of the deletion. | ||
| */ | ||
| abstract public void clean(long threshTime, boolean waitForCompletion); | ||
|
|
||
| /** | ||
| * Close this log and release any resources. | ||
| */ | ||
| abstract public void close(); | ||
| } |
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,30 @@ | ||
| /* | ||
| * Licensed to the Apache Software Foundation (ASF) under one or more | ||
| * contributor license agreements. See the NOTICE file distributed with | ||
| * this work for additional information regarding copyright ownership. | ||
| * The ASF licenses this file to You under the Apache License, Version 2.0 | ||
| * (the "License"); you may not use this file except in compliance with | ||
| * the License. You may obtain a copy of the License at | ||
| * | ||
| * http://www.apache.org/licenses/LICENSE-2.0 | ||
| * | ||
| * Unless required by applicable law or agreed to in writing, software | ||
| * distributed under the License is distributed on an "AS IS" BASIS, | ||
| * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. | ||
| * See the License for the specific language governing permissions and | ||
| * limitations under the License. | ||
| */ | ||
|
|
||
| package org.apache.spark.streaming.util; | ||
|
|
||
| /** | ||
| * This abstract class represents a handle that refers to a record written in a | ||
| * {@link org.apache.spark.streaming.util.WriteAheadLog WriteAheadLog}. | ||
| * It must contain all the information necessary for the record to be read and returned by | ||
| * an implemenation of the WriteAheadLog class. | ||
| * | ||
| * @see org.apache.spark.streaming.util.WriteAheadLog | ||
| */ | ||
| @org.apache.spark.annotation.DeveloperApi | ||
| public abstract class WriteAheadLogRecordHandle implements java.io.Serializable { | ||
| } |
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -16,14 +16,17 @@ | |
| */ | ||
| package org.apache.spark.streaming.rdd | ||
|
|
||
| import java.nio.ByteBuffer | ||
|
|
||
| import scala.reflect.ClassTag | ||
| import scala.util.control.NonFatal | ||
|
|
||
| import org.apache.commons.io.FileUtils | ||
|
|
||
| import org.apache.spark._ | ||
| import org.apache.spark.rdd.BlockRDD | ||
| import org.apache.spark.storage.{BlockId, StorageLevel} | ||
| import org.apache.spark.streaming.util._ | ||
|
|
||
| /** | ||
| * Partition class for [[org.apache.spark.streaming.rdd.WriteAheadLogBackedBlockRDD]]. | ||
|
|
@@ -37,7 +40,7 @@ private[streaming] | |
| class WriteAheadLogBackedBlockRDDPartition( | ||
| val index: Int, | ||
| val blockId: BlockId, | ||
| val segment: WriteAheadLogRecordHandle) | ||
| extends Partition | ||
|
|
||
|
|
||
|
|
@@ -58,7 +61,7 @@ private[streaming] | |
| class WriteAheadLogBackedBlockRDD[T: ClassTag]( | ||
| @transient sc: SparkContext, | ||
| @transient blockIds: Array[BlockId], | ||
| @transient segments: Array[WriteAheadLogRecordHandle], | ||
| storeInBlockManager: Boolean, | ||
| storageLevel: StorageLevel) | ||
| extends BlockRDD[T](sc, blockIds) { | ||
|
|
@@ -96,9 +99,27 @@ class WriteAheadLogBackedBlockRDD[T: ClassTag]( | |
| logDebug(s"Read partition data of $this from block manager, block $blockId") | ||
| iterator | ||
| case None => // Data not found in Block Manager, grab it from write ahead log file | ||
| var dataRead: ByteBuffer = null | ||
| var writeAheadLog: WriteAheadLog = null | ||
| try { | ||
| val dummyDirectory = FileUtils.getTempDirectoryPath() | ||
|
Contributor
There was a problem hiding this comment. Why here need to use
Contributor
Author
There was a problem hiding this comment. So the default WAL is file based so a log directory is needed for it to work. However, the log directory is really not needed reading a particular record. But to read a single record you have to create a FileBasedWriteAheadLog object, which needs a log directory. Hence I am providing a dummy directory for this. I know that this is a little awkward. This is the cost of defining a single interface for both writing and reading single records. Earlier there were two independent classes (WALWriter and WALRandomReader) that was used for these two purposes, which has different requirements. But since I am trying make single interface that can be used for all reading and writing, the log directory must be provided in the constructor of the default file-based WAL. This results in the awkwardness. I dont quite like it myself, but it may practically be okay as long as we ensure that the FileBasedWAL does not create unnecessary directories/files when only reading a single record. I can add a test to ensure that. |
||
| writeAheadLog = WriteAheadLogUtils.createLogForReceiver( | ||
| SparkEnv.get.conf, dummyDirectory, hadoopConf) | ||
|
Contributor
There was a problem hiding this comment. Also IIUC here if the journal system if not hadoop based, hadoopConf may not be available.
Contributor
Author
There was a problem hiding this comment. hadoopConf is always available through the SparkContext. Irrespective of whether Hadoop file system is used, a Hadoop conf is created by the SparkContext which is passed on to this location. If the WAL is not the default FileBasedWAL, then this parameter is just ignored (see the method
Contributor
There was a problem hiding this comment. What I'm thinking is that do we need to have this parameter for the interface, can we hide this into file-based WAL implementations.
Contributor
Author
There was a problem hiding this comment. The log directory needs to be passed through the IMO that duplicates code everywhere and uglier that this dummy dir approach. And also, this does not handle |
||
| dataRead = writeAheadLog.read(partition.segment) | ||
| } catch { | ||
| case NonFatal(e) => | ||
| throw new SparkException( | ||
| s"Could not read data from write ahead log segment ${partition.segment}", e) | ||
| } finally { | ||
| if (writeAheadLog != null) { | ||
| writeAheadLog.close() | ||
|
Contributor
There was a problem hiding this comment. May be reset writeAheadLog to null after close to avoid unexpected behavior :).
|
||
| } | ||
| } | ||
| if (dataRead == null) { | ||
| throw new SparkException( | ||
| s"Could not read data from write ahead log segment ${partition.segment}, " + | ||
| s"read returned null") | ||
| } | ||
| logInfo(s"Read partition data of $this from write ahead log, segment ${partition.segment}") | ||
| if (storeInBlockManager) { | ||
| blockManager.putBytes(blockId, dataRead, storageLevel) | ||
|
|
@@ -117,8 +138,14 @@ class WriteAheadLogBackedBlockRDD[T: ClassTag]( | |
| override def getPreferredLocations(split: Partition): Seq[String] = { | ||
| val partition = split.asInstanceOf[WriteAheadLogBackedBlockRDDPartition] | ||
| val blockLocations = getBlockIdLocations().get(partition.blockId) | ||
| blockLocations.getOrElse { | ||
|
Contributor
There was a problem hiding this comment. It might make sense to add location info to the WALRecordHandle interface itself. This way, systems that are not HDFS, but still benefit from preferred locations can use it.
Contributor
Author
There was a problem hiding this comment. That's a good point. I wasnt super sure of whether it is a good idea to have it in the interface in this version. We can add it later and maintain binary compatibility as the RecordHandle is an abstract class. Also It is still a developer API s. For now, I am going to merge this in to unblock #5732 . |
||
| partition.segment match { | ||
| case fileSegment: FileBasedWriteAheadLogSegment => | ||
| HdfsUtils.getFileSegmentLocations( | ||
| fileSegment.path, fileSegment.offset, fileSegment.length, hadoopConfig) | ||
| case _ => | ||
| Seq.empty | ||
| } | ||
| } | ||
| } | ||
| } | ||
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -17,18 +17,18 @@ | |
|
|
||
| package org.apache.spark.streaming.receiver | ||
|
|
||
| import scala.concurrent.duration._ | ||
| import scala.concurrent.{Await, ExecutionContext, Future} | ||
| import scala.language.{existentials, postfixOps} | ||
|
|
||
| import org.apache.hadoop.conf.Configuration | ||
| import org.apache.hadoop.fs.Path | ||
|
|
||
| import org.apache.spark.storage._ | ||
| import org.apache.spark.streaming.receiver.WriteAheadLogBasedBlockHandler._ | ||
| import org.apache.spark.streaming.util.{WriteAheadLogRecordHandle, WriteAheadLogUtils} | ||
| import org.apache.spark.util.{Clock, SystemClock, ThreadUtils} | ||
| import org.apache.spark.{Logging, SparkConf, SparkException} | ||
|
|
||
| /** Trait that represents the metadata related to storage of blocks */ | ||
| private[streaming] trait ReceivedBlockStoreResult { | ||
|
|
@@ -96,7 +96,7 @@ private[streaming] class BlockManagerBasedBlockHandler( | |
| */ | ||
| private[streaming] case class WriteAheadLogBasedStoreResult( | ||
| blockId: StreamBlockId, | ||
| segment: WriteAheadLogRecordHandle | ||
|
Contributor
Author
There was a problem hiding this comment. This is a leftover from the name change, segment --> handle. If there is no other change needed, I can take care of it in the next PR #5732
Contributor
Author
There was a problem hiding this comment. On second thought, I am fixing this. This is a problem in too many places. |
||
| ) extends ReceivedBlockStoreResult | ||
|
|
||
|
|
||
|
|
@@ -116,10 +116,6 @@ private[streaming] class WriteAheadLogBasedBlockHandler( | |
|
|
||
| private val blockStoreTimeout = conf.getInt( | ||
| "spark.streaming.receiver.blockStoreTimeout", 30).seconds | ||
|
|
||
| private val effectiveStorageLevel = { | ||
| if (storageLevel.deserialized) { | ||
|
|
@@ -139,13 +135,9 @@ private[streaming] class WriteAheadLogBasedBlockHandler( | |
| s"$effectiveStorageLevel when write ahead log is enabled") | ||
| } | ||
|
|
||
| // Write ahead log manages | ||
| private val writeAheadLog = WriteAheadLogUtils.createLogForReceiver( | ||
| conf, checkpointDirToLogDir(checkpointDir, streamId), hadoopConf) | ||
|
|
||
| // For processing futures used in parallel block storing into block manager and write ahead log | ||
| // # threads = 2, so that both writing to BM and WAL can proceed in parallel | ||
|
|
@@ -183,7 +175,7 @@ private[streaming] class WriteAheadLogBasedBlockHandler( | |
|
|
||
| // Store the block in write ahead log | ||
| val storeInWriteAheadLogFuture = Future { | ||
| writeAheadLog.write(serializedBlock, clock.getTimeMillis()) | ||
| } | ||
|
|
||
| // Combine the futures, wait for both to complete, and return the write ahead log segment | ||
|
|
@@ -193,11 +185,11 @@ private[streaming] class WriteAheadLogBasedBlockHandler( | |
| } | ||
|
|
||
| def cleanupOldBlocks(threshTime: Long) { | ||
| writeAheadLog.clean(threshTime, false) | ||
| } | ||
|
|
||
| def stop() { | ||
| writeAheadLog.close() | ||
| } | ||
| } | ||
|
|
||
|
|
||
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
I feel dirty seeing nulls in scala
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Why allocate two (at least) objects when it is completely obvious that they are not going to be used. The null does not get exposed to anything outside the function, and hence is okay to have.
If you look at rest of the Spark source code, we dont strictly adhere to Scala-way of doing things, rather balance code understandability (limit the levels of functional nesting) and efficiency (while loops instead of for when perf matters) with Scala styles.