-
- Downloads
[SPARK-7056] [STREAMING] Make the Write Ahead Log pluggable
Users may want the WAL data to be written to non-HDFS data storage systems. To allow that, we have to make the WAL pluggable. The following design doc outlines the plan. https://docs.google.com/a/databricks.com/document/d/1A2XaOLRFzvIZSi18i_luNw5Rmm9j2j4AigktXxIYxmY/edit?usp=sharing Things to add. * Unit tests for WriteAheadLogUtils Author: Tathagata Das <tathagata.das1565@gmail.com> Closes #5645 from tdas/wal-pluggable and squashes the following commits: 2c431fd [Tathagata Das] Minor fixes. c2bc7384 [Tathagata Das] More changes based on PR comments. 569a416 [Tathagata Das] fixed long line bde26b1 [Tathagata Das] Renamed segment to record handle everywhere b65e155 [Tathagata Das] More changes based on PR comments. d7cd15b [Tathagata Das] Fixed test 1a32a4b [Tathagata Das] Fixed test e0d19fb [Tathagata Das] Fixed defaults 9310cbf [Tathagata Das] style fix. 86abcb1 [Tathagata Das] Refactored WriteAheadLogUtils, and consolidated all WAL related configuration into it. 84ce469 [Tathagata Das] Added unit test and fixed compilation error. bce5e75 [Tathagata Das] Fixed long lines. 837c4f5 [Tathagata Das] Merge remote-tracking branch 'apache-github/master' into wal-pluggable 754fbf8 [Tathagata Das] Added license and docs. 09bc6fe [Tathagata Das] Merge remote-tracking branch 'apache-github/master' into wal-pluggable 7dd2d4b [Tathagata Das] Added pluggable WriteAheadLog interface, and refactored all code along with it
Showing
- external/kafka/src/main/scala/org/apache/spark/streaming/kafka/KafkaUtils.scala 2 additions, 1 deletion...n/scala/org/apache/spark/streaming/kafka/KafkaUtils.scala
- streaming/src/main/java/org/apache/spark/streaming/util/WriteAheadLog.java 60 additions, 0 deletions...n/java/org/apache/spark/streaming/util/WriteAheadLog.java
- streaming/src/main/java/org/apache/spark/streaming/util/WriteAheadLogRecordHandle.java 30 additions, 0 deletions...pache/spark/streaming/util/WriteAheadLogRecordHandle.java
- streaming/src/main/scala/org/apache/spark/streaming/dstream/ReceiverInputDStream.scala 1 addition, 1 deletion...apache/spark/streaming/dstream/ReceiverInputDStream.scala
- streaming/src/main/scala/org/apache/spark/streaming/rdd/WriteAheadLogBackedBlockRDD.scala 57 additions, 22 deletions...che/spark/streaming/rdd/WriteAheadLogBackedBlockRDD.scala
- streaming/src/main/scala/org/apache/spark/streaming/receiver/ReceivedBlockHandler.scala 15 additions, 23 deletions...pache/spark/streaming/receiver/ReceivedBlockHandler.scala
- streaming/src/main/scala/org/apache/spark/streaming/receiver/ReceiverSupervisorImpl.scala 3 additions, 2 deletions...che/spark/streaming/receiver/ReceiverSupervisorImpl.scala
- streaming/src/main/scala/org/apache/spark/streaming/scheduler/ReceivedBlockTracker.scala 18 additions, 20 deletions...ache/spark/streaming/scheduler/ReceivedBlockTracker.scala
- streaming/src/main/scala/org/apache/spark/streaming/scheduler/ReceiverTracker.scala 2 additions, 1 deletion...rg/apache/spark/streaming/scheduler/ReceiverTracker.scala
- streaming/src/main/scala/org/apache/spark/streaming/util/FileBasedWriteAheadLog.scala 46 additions, 30 deletions.../apache/spark/streaming/util/FileBasedWriteAheadLog.scala
- streaming/src/main/scala/org/apache/spark/streaming/util/FileBasedWriteAheadLogRandomReader.scala 4 additions, 4 deletions...k/streaming/util/FileBasedWriteAheadLogRandomReader.scala
- streaming/src/main/scala/org/apache/spark/streaming/util/FileBasedWriteAheadLogReader.scala 2 additions, 2 deletions...e/spark/streaming/util/FileBasedWriteAheadLogReader.scala
- streaming/src/main/scala/org/apache/spark/streaming/util/FileBasedWriteAheadLogSegment.scala 2 additions, 1 deletion.../spark/streaming/util/FileBasedWriteAheadLogSegment.scala
- streaming/src/main/scala/org/apache/spark/streaming/util/FileBasedWriteAheadLogWriter.scala 4 additions, 5 deletions...e/spark/streaming/util/FileBasedWriteAheadLogWriter.scala
- streaming/src/main/scala/org/apache/spark/streaming/util/WriteAheadLogUtils.scala 129 additions, 0 deletions.../org/apache/spark/streaming/util/WriteAheadLogUtils.scala
- streaming/src/test/java/org/apache/spark/streaming/JavaWriteAheadLogSuite.java 129 additions, 0 deletions...va/org/apache/spark/streaming/JavaWriteAheadLogSuite.java
- streaming/src/test/scala/org/apache/spark/streaming/ReceivedBlockHandlerSuite.scala 11 additions, 7 deletions...rg/apache/spark/streaming/ReceivedBlockHandlerSuite.scala
- streaming/src/test/scala/org/apache/spark/streaming/ReceivedBlockTrackerSuite.scala 16 additions, 12 deletions...rg/apache/spark/streaming/ReceivedBlockTrackerSuite.scala
- streaming/src/test/scala/org/apache/spark/streaming/ReceiverSuite.scala 1 addition, 1 deletion...test/scala/org/apache/spark/streaming/ReceiverSuite.scala
- streaming/src/test/scala/org/apache/spark/streaming/StreamingContextSuite.scala 2 additions, 2 deletions...la/org/apache/spark/streaming/StreamingContextSuite.scala
Loading
Please register or sign in to comment