Skip to content

Commit c25ca7c

Browse files
emressrowen
authored andcommitted
SPARK-3276 Added a new configuration spark.streaming.minRememberDuration
SPARK-3276 Added a new configuration parameter ``spark.streaming.minRememberDuration``, with a default value of 1 minute. So that when a Spark Streaming application is started, an arbitrary number of minutes can be taken as threshold for remembering. Author: emres <[email protected]> Closes #5438 from emres/SPARK-3276 and squashes the following commits: 766f938 [emres] SPARK-3276 Switched to using newly added getTimeAsSeconds method. affee1d [emres] SPARK-3276 Changed the property name and variable name for minRememberDuration c9d58ca [emres] SPARK-3276 Minor code re-formatting. 1c53ba9 [emres] SPARK-3276 Started to use ssc.conf rather than ssc.sparkContext.getConf, and also getLong method directly. bfe0acb [emres] SPARK-3276 Moved the minRememberDurationMin to the class daccc82 [emres] SPARK-3276 Changed the property name to reflect the unit of value and reduced number of fields. 43cc1ce [emres] SPARK-3276 Added a new configuration parameter spark.streaming.minRemember duration, with a default value of 1 minute.
1 parent c035c0f commit c25ca7c

File tree

1 file changed

+17
-13
lines changed

1 file changed

+17
-13
lines changed

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

Lines changed: 17 additions & 13 deletions
Original file line numberDiff line numberDiff line change
@@ -26,7 +26,7 @@ import org.apache.hadoop.conf.Configuration
2626
import org.apache.hadoop.fs.{FileSystem, Path, PathFilter}
2727
import org.apache.hadoop.mapreduce.{InputFormat => NewInputFormat}
2828

29-
import org.apache.spark.SerializableWritable
29+
import org.apache.spark.{SparkConf, SerializableWritable}
3030
import org.apache.spark.rdd.{RDD, UnionRDD}
3131
import org.apache.spark.streaming._
3232
import org.apache.spark.util.{TimeStampedHashMap, Utils}
@@ -63,7 +63,7 @@ import org.apache.spark.util.{TimeStampedHashMap, Utils}
6363
* the streaming app.
6464
* - If a file is to be visible in the directory listings, it must be visible within a certain
6565
* duration of the mod time of the file. This duration is the "remember window", which is set to
66-
* 1 minute (see `FileInputDStream.MIN_REMEMBER_DURATION`). Otherwise, the file will never be
66+
* 1 minute (see `FileInputDStream.minRememberDuration`). Otherwise, the file will never be
6767
* selected as the mod time will be less than the ignore threshold when it becomes visible.
6868
* - Once a file is visible, the mod time cannot change. If it does due to appends, then the
6969
* processing semantics are undefined.
@@ -80,6 +80,15 @@ class FileInputDStream[K, V, F <: NewInputFormat[K,V]](
8080

8181
private val serializableConfOpt = conf.map(new SerializableWritable(_))
8282

83+
/**
84+
* Minimum duration of remembering the information of selected files. Defaults to 60 seconds.
85+
*
86+
* Files with mod times older than this "window" of remembering will be ignored. So if new
87+
* files are visible within this window, then the file will get selected in the next batch.
88+
*/
89+
private val minRememberDurationS =
90+
Seconds(ssc.conf.getTimeAsSeconds("spark.streaming.minRememberDuration", "60s"))
91+
8392
// This is a def so that it works during checkpoint recovery:
8493
private def clock = ssc.scheduler.clock
8594

@@ -95,7 +104,8 @@ class FileInputDStream[K, V, F <: NewInputFormat[K,V]](
95104
* This would allow us to filter away not-too-old files which have already been recently
96105
* selected and processed.
97106
*/
98-
private val numBatchesToRemember = FileInputDStream.calculateNumBatchesToRemember(slideDuration)
107+
private val numBatchesToRemember = FileInputDStream
108+
.calculateNumBatchesToRemember(slideDuration, minRememberDurationS)
99109
private val durationToRemember = slideDuration * numBatchesToRemember
100110
remember(durationToRemember)
101111

@@ -330,20 +340,14 @@ class FileInputDStream[K, V, F <: NewInputFormat[K,V]](
330340
private[streaming]
331341
object FileInputDStream {
332342

333-
/**
334-
* Minimum duration of remembering the information of selected files. Files with mod times
335-
* older than this "window" of remembering will be ignored. So if new files are visible
336-
* within this window, then the file will get selected in the next batch.
337-
*/
338-
private val MIN_REMEMBER_DURATION = Minutes(1)
339-
340343
def defaultFilter(path: Path): Boolean = !path.getName().startsWith(".")
341344

342345
/**
343346
* Calculate the number of last batches to remember, such that all the files selected in
344-
* at least last MIN_REMEMBER_DURATION duration can be remembered.
347+
* at least last minRememberDurationS duration can be remembered.
345348
*/
346-
def calculateNumBatchesToRemember(batchDuration: Duration): Int = {
347-
math.ceil(MIN_REMEMBER_DURATION.milliseconds.toDouble / batchDuration.milliseconds).toInt
349+
def calculateNumBatchesToRemember(batchDuration: Duration,
350+
minRememberDurationS: Duration): Int = {
351+
math.ceil(minRememberDurationS.milliseconds.toDouble / batchDuration.milliseconds).toInt
348352
}
349353
}

0 commit comments

Comments
 (0)