From 60e9b2bdd55854a6be077b17aa032c25bfb031bf Mon Sep 17 00:00:00 2001 From: shaofei007 <1427357147@qq.com> Date: Sat, 29 Jul 2017 13:27:39 +0100 Subject: [PATCH] [SPARK-21357][DSTREAMS] FileInputDStream not remove out of date RDD ## What changes were proposed in this pull request? ```DStreams class FileInputDStream [line 162] protected[streaming] override def clearMetadata(time: Time) { batchTimeToSelectedFiles.synchronized { val oldFiles = batchTimeToSelectedFiles.filter(_._1 < (time - rememberDuration)) batchTimeToSelectedFiles --= oldFiles.keys ``` The above code does not remove the old generatedRDDs. "super.clearMetadata(time)" was added to the beginning of clearMetadata to remove the old generatedRDDs. ## How was this patch tested? At the end of clearMetadata, the testing code (print the number of generatedRDDs) was added to check the old RDDS were removed manually. Author: shaofei007 <1427357147@qq.com> Author: Fei Shao <1427357147@qq.com> Closes #18718 from shaofei007/master. --- .../org/apache/spark/streaming/dstream/FileInputDStream.scala | 1 + 1 file changed, 1 insertion(+) diff --git a/streaming/src/main/scala/org/apache/spark/streaming/dstream/FileInputDStream.scala b/streaming/src/main/scala/org/apache/spark/streaming/dstream/FileInputDStream.scala index 905b1c52af..b8a5a96faf 100644 --- a/streaming/src/main/scala/org/apache/spark/streaming/dstream/FileInputDStream.scala +++ b/streaming/src/main/scala/org/apache/spark/streaming/dstream/FileInputDStream.scala @@ -164,6 +164,7 @@ class FileInputDStream[K, V, F <: NewInputFormat[K, V]]( /** Clear the old time-to-files mappings along with old RDDs */ protected[streaming] override def clearMetadata(time: Time) { + super.clearMetadata(time) batchTimeToSelectedFiles.synchronized { val oldFiles = batchTimeToSelectedFiles.filter(_._1 < (time - rememberDuration)) batchTimeToSelectedFiles --= oldFiles.keys -- GitLab