[SPARK-7139][Streaming] Allow received block metadata to be saved to WAL and recovered on driver failure #5732

tdas · 2015-04-28T03:35:53Z

Enabled ReceivedBlockTracker WAL by default
Stored block metadata in the WAL
Optimized WALBackedBlockRDD by skipping block fetch when the block is known to not exist in Spark

…timized block fetching in WALBackedBlockRDD

tdas · 2015-04-28T03:36:26Z

@harishreedharan @jerryshao Take a look.
@pwendell Take a look at the BlockRDD as it touches core.

SparkQA · 2015-04-28T04:52:54Z

Test build #31106 has finished for PR 5732 at commit d06fa21.

This patch fails Spark unit tests.
This patch merges cleanly.
This patch adds no public classes.
This patch does not change any dependencies.

SparkQA · 2015-04-28T09:24:54Z

Test build #31135 has finished for PR 5732 at commit 1bc5bc3.

This patch fails Spark unit tests.
This patch merges cleanly.
This patch adds no public classes.
This patch does not change any dependencies.

SparkQA · 2015-04-29T10:30:46Z

Test build #31266 has finished for PR 5732 at commit 5f67a59.

This patch fails PySpark unit tests.
This patch merges cleanly.
This patch adds no public classes.
This patch adds the following new dependencies:
- spark-unsafe_2.10-1.4.0-SNAPSHOT.jar

Conflicts: streaming/src/main/scala/org/apache/spark/streaming/dstream/ReceiverInputDStream.scala streaming/src/main/scala/org/apache/spark/streaming/rdd/WriteAheadLogBackedBlockRDD.scala streaming/src/main/scala/org/apache/spark/streaming/receiver/ReceiverSupervisorImpl.scala streaming/src/main/scala/org/apache/spark/streaming/scheduler/ReceivedBlockTracker.scala streaming/src/test/scala/org/apache/spark/streaming/ReceivedBlockTrackerSuite.scala streaming/src/test/scala/org/apache/spark/streaming/rdd/WriteAheadLogBackedBlockRDDSuite.scala

tdas · 2015-04-30T01:42:34Z

@jerryshao @harishreedharan Please take a look at this PR.

tdas · 2015-04-30T01:44:58Z

streaming/src/main/scala/org/apache/spark/streaming/rdd/WriteAheadLogBackedBlockRDD.scala

-        }
-        if (dataRead == null) {
+
+    def getBlockFromBlockManager(): Option[Iterator[T]] = {


This whole thing just puts the old code within inner functions, not really a big change. The real change is that these functions are called selectively in lines 157..160.

SparkQA · 2015-04-30T02:26:08Z

Test build #31360 has finished for PR 5732 at commit 466212c.

This patch passes all tests.
This patch merges cleanly.
This patch adds no public classes.
This patch does not change any dependencies.

jerryshao · 2015-04-30T13:42:14Z

streaming/src/main/scala/org/apache/spark/streaming/dstream/ReceiverInputDStream.scala

-          logWarning("Multiple result types in block information, WAL information will be ignored.")
-        }
+        // Is WAL segment info present with all the blocks
+        val isWALSegmentInfoPresent = blockInfos.forall { _.walRecordHandleOption.nonEmpty }


is it better to rename as isWALRecordHandleInfoPresent ? Looked though this part of code, some variables use segment but actually is handle, might be better to keep consistent :)

SparkQA · 2015-04-30T21:44:47Z

Test build #31446 has finished for PR 5732 at commit 637bc9c.

This patch passes all tests.
This patch merges cleanly.
This patch adds no public classes.
This patch does not change any dependencies.

tdas · 2015-05-01T03:13:04Z

@harishreedharan Any thoughts?

harishreedharan · 2015-05-01T03:37:37Z

streaming/src/main/scala/org/apache/spark/streaming/rdd/WriteAheadLogBackedBlockRDD.scala

@@ -40,21 +40,30 @@ private[streaming]
 class WriteAheadLogBackedBlockRDDPartition(
    val index: Int,
    val blockId: BlockId,
-    val walRecordHandle: WriteAheadLogRecordHandle)
-  extends Partition
+    val isBlockIdValid: Boolean,


Can you add this param to the scaladoc?

Good catch.

harishreedharan · 2015-05-01T03:39:44Z

LGTM.

pwendell · 2015-05-01T22:00:33Z

core/src/main/scala/org/apache/spark/rdd/BlockRDD.scala

@@ -66,7 +67,9 @@ class BlockRDD[T: ClassTag](@transient sc: SparkContext, @transient val blockIds
    blockIds.foreach { blockId =>
      sc.env.blockManager.master.removeBlock(blockId)
    }
-    _isValid = false
+    if (_setInvalid) {


So rather than having this maybe the subclass of this could override isValid. It's a bit confusing as it stands.

Will do. That is a more intuitive design.

Also, I will add unit tests to test this behavior for BlockRDD and WALBackedBlockRDD.

Conflicts: streaming/src/main/scala/org/apache/spark/streaming/dstream/ReceiverInputDStream.scala streaming/src/main/scala/org/apache/spark/streaming/rdd/WriteAheadLogBackedBlockRDD.scala streaming/src/test/scala/org/apache/spark/streaming/ui/StreamingJobProgressListenerSuite.scala

tdas · 2015-05-05T02:31:03Z

I addressed @pwendell comments as well as added more unit tests to WriteAheadLogBackedBlockRDDSuite to get 100% code coverage over WriteAheadLogBackedBlockRDD. Also got 100% coverage of ReceivedBlockTracker.

tdas · 2015-05-05T02:31:11Z

I will merge this when tests pass.

SparkQA · 2015-05-05T03:48:02Z

Test build #31817 has finished for PR 5732 at commit 575476e.

This patch fails Spark unit tests.
This patch merges cleanly.
This patch adds no public classes.

tdas · 2015-05-05T05:18:04Z

retest this please.

SparkQA · 2015-05-05T07:05:23Z

Test build #31836 has finished for PR 5732 at commit 575476e.

This patch passes all tests.
This patch merges cleanly.
This patch adds no public classes.

… WAL and recovered on driver failure - Enabled ReceivedBlockTracker WAL by default - Stored block metadata in the WAL - Optimized WALBackedBlockRDD by skipping block fetch when the block is known to not exist in Spark Author: Tathagata Das <tathagata.das1565@gmail.com> Closes #5732 from tdas/SPARK-7139 and squashes the following commits: 575476e [Tathagata Das] Added more tests to get 100% coverage of the WALBackedBlockRDD 19668ba [Tathagata Das] Merge remote-tracking branch 'apache-github/master' into SPARK-7139 685fab3 [Tathagata Das] Addressed comments in PR 637bc9c [Tathagata Das] Changed segment to handle 466212c [Tathagata Das] Merge remote-tracking branch 'apache-github/master' into SPARK-7139 5f67a59 [Tathagata Das] Fixed HdfsUtils to handle append in local file system 1bc5bc3 [Tathagata Das] Fixed bug on unexpected recovery d06fa21 [Tathagata Das] Enabled ReceivedBlockTracker by default, stored block metadata and optimized block fetching in WALBackedBlockRDD (cherry picked from commit 1854ac3) Signed-off-by: Tathagata Das <tathagata.das1565@gmail.com>

… WAL and recovered on driver failure - Enabled ReceivedBlockTracker WAL by default - Stored block metadata in the WAL - Optimized WALBackedBlockRDD by skipping block fetch when the block is known to not exist in Spark Author: Tathagata Das <tathagata.das1565@gmail.com> Closes apache#5732 from tdas/SPARK-7139 and squashes the following commits: 575476e [Tathagata Das] Added more tests to get 100% coverage of the WALBackedBlockRDD 19668ba [Tathagata Das] Merge remote-tracking branch 'apache-github/master' into SPARK-7139 685fab3 [Tathagata Das] Addressed comments in PR 637bc9c [Tathagata Das] Changed segment to handle 466212c [Tathagata Das] Merge remote-tracking branch 'apache-github/master' into SPARK-7139 5f67a59 [Tathagata Das] Fixed HdfsUtils to handle append in local file system 1bc5bc3 [Tathagata Das] Fixed bug on unexpected recovery d06fa21 [Tathagata Das] Enabled ReceivedBlockTracker by default, stored block metadata and optimized block fetching in WALBackedBlockRDD

Enabled ReceivedBlockTracker by default, stored block metadata and op…

d06fa21

…timized block fetching in WALBackedBlockRDD

Fixed bug on unexpected recovery

1bc5bc3

This was referenced Apr 28, 2015

[SPARK-6862][Streaming][WebUI] Add BatchPage to display details of a batch #5473

Closed

[SPARK-7056][Streaming] Make the Write Ahead Log pluggable #5645

Closed

Fixed HdfsUtils to handle append in local file system

5f67a59

tdas reviewed Apr 30, 2015
View reviewed changes

jerryshao reviewed Apr 30, 2015
View reviewed changes

Changed segment to handle

637bc9c

harishreedharan reviewed May 1, 2015
View reviewed changes

pwendell reviewed May 1, 2015
View reviewed changes

tdas added 3 commits May 2, 2015 02:16

Addressed comments in PR

685fab3

Added more tests to get 100% coverage of the WALBackedBlockRDD

575476e

asfgit closed this in 1854ac3 May 5, 2015

jerryshao mentioned this pull request May 6, 2015

[SPARK-7397][Streaming] Add missing inputInfo report code #5937

Closed

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

[SPARK-7139][Streaming] Allow received block metadata to be saved to WAL and recovered on driver failure #5732

[SPARK-7139][Streaming] Allow received block metadata to be saved to WAL and recovered on driver failure #5732

tdas commented Apr 28, 2015

tdas commented Apr 28, 2015

SparkQA commented Apr 28, 2015

SparkQA commented Apr 28, 2015

SparkQA commented Apr 29, 2015

tdas commented Apr 30, 2015

tdas Apr 30, 2015

SparkQA commented Apr 30, 2015

jerryshao Apr 30, 2015

tdas Apr 30, 2015

SparkQA commented Apr 30, 2015

tdas commented May 1, 2015

harishreedharan May 1, 2015

tdas May 1, 2015

harishreedharan commented May 1, 2015

pwendell May 1, 2015

tdas May 1, 2015

tdas May 1, 2015

tdas commented May 5, 2015

tdas commented May 5, 2015

SparkQA commented May 5, 2015

tdas commented May 5, 2015

SparkQA commented May 5, 2015

[SPARK-7139][Streaming] Allow received block metadata to be saved to WAL and recovered on driver failure #5732

[SPARK-7139][Streaming] Allow received block metadata to be saved to WAL and recovered on driver failure #5732

Conversation

tdas commented Apr 28, 2015

tdas commented Apr 28, 2015

SparkQA commented Apr 28, 2015

SparkQA commented Apr 28, 2015

SparkQA commented Apr 29, 2015

tdas commented Apr 30, 2015

Choose a reason for hiding this comment

SparkQA commented Apr 30, 2015

Choose a reason for hiding this comment

Choose a reason for hiding this comment

SparkQA commented Apr 30, 2015

tdas commented May 1, 2015

Choose a reason for hiding this comment

Choose a reason for hiding this comment

harishreedharan commented May 1, 2015

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

tdas commented May 5, 2015

tdas commented May 5, 2015

SparkQA commented May 5, 2015

tdas commented May 5, 2015

SparkQA commented May 5, 2015