Skip to content

SPARK-3276 Added a new configuration spark.streaming.minRememberDuration #5438

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
wants to merge 7 commits into from
Closed

SPARK-3276 Added a new configuration spark.streaming.minRememberDuration #5438

Changes from 1 commit
Commits
Show all changes
7 commits
Select commit Hold shift + click to select a range
43cc1ce
SPARK-3276 Added a new configuration parameter spark.streaming.minRem…
emres Apr 9, 2015
daccc82
SPARK-3276 Changed the property name to reflect the unit of value and…
emres Apr 9, 2015
bfe0acb
SPARK-3276 Moved the minRememberDurationMin to the class
emres Apr 13, 2015
1c53ba9
SPARK-3276 Started to use ssc.conf rather than ssc.sparkContext.getCo…
emres Apr 13, 2015
c9d58ca
SPARK-3276 Minor code re-formatting.
emres Apr 13, 2015
affee1d
SPARK-3276 Changed the property name and variable name for minRemembe…
emres Apr 13, 2015
766f938
SPARK-3276 Switched to using newly added getTimeAsSeconds method.
emres Apr 14, 2015
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
12 changes: 7 additions & 5 deletions streaming/src/main/scala/org/apache/spark/streaming/dstream/FileInputDStream.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -26,7 +26,7 @@ import org.apache.hadoop.conf.Configuration
import org.apache.hadoop.fs.{FileSystem, Path, PathFilter}
import org.apache.hadoop.mapreduce.{InputFormat => NewInputFormat}

import org.apache.spark.SerializableWritable
import org.apache.spark.{SparkConf, SerializableWritable}
import org.apache.spark.rdd.{RDD, UnionRDD}
import org.apache.spark.streaming._
import org.apache.spark.util.{TimeStampedHashMap, Utils}
Expand Down Expand Up @@ -331,11 +331,13 @@ private[streaming]
object FileInputDStream {

/**
* Minimum duration of remembering the information of selected files. Files with mod times
* older than this "window" of remembering will be ignored. So if new files are visible
* within this window, then the file will get selected in the next batch.
* Minimum duration of remembering the information of selected files. Defaults to 1 minute.
*
* Files with mod times older than this "window" of remembering will be ignored. So if new
* files are visible within this window, then the file will get selected in the next batch.
*/
private val MIN_REMEMBER_DURATION = Minutes(1)
private val minRememberDuration = new SparkConf().get("spark.streaming.minRememberDuration", "1")
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Note https://issues.apache.org/jira/browse/SPARK-5931 here which would update time properties to be specifiable like "3s". It isn't updating property names, but we could update a new one. I suggest this property name include the units.

It looks like it's minutes here, so spark.streaming.minRememberDurationMin. Can it be seconds to allow finer granularity or is that unreasonable?

Why two fields here? just one minRememberDurationMin (or Sec) is needed?

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

@srowen changing the property name to spark.streaming.minRememberDurationMin makes sense with regard to consistency in naming. I will also get rid of of two fields. I'll open another pull request that reflects those changes.

private val MIN_REMEMBER_DURATION = Minutes(minRememberDuration.toLong)

def defaultFilter(path: Path): Boolean = !path.getName().startsWith(".")

Expand Down