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
emres 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
Prev Previous commit
Next Next commit
SPARK-3276 Started to use ssc.conf rather than ssc.sparkContext.getCo… 
…nf, and also getLong method directly.
  • Loading branch information
@emres
emres committed Apr 13, 2015
commit 1c53ba9ed84b2e2a5129f4cedd0574060704c9d7
5 changes: 2 additions & 3 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 @@ -86,9 +86,8 @@ class FileInputDStream[K, V, F <: NewInputFormat[K,V]](
* 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 minRememberDurationMin = Minutes(ssc.sparkContext.getConf
.get("spark.streaming.minRememberDurationMin", "1")
.toLong)
private val minRememberDurationMin = Minutes(ssc.conf
Copy link
Member

@srowen srowen Apr 13, 2015

Choose a reason for hiding this comment

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

The continuation indent looks a bit off here. Maybe just start the next line with Minutes(....
I think we might want to put this change on hold until SPARK-5931 is resolved, since it will give some proper syntax for time-based properties that you can use here.

.getLong("spark.streaming.minRememberDurationMin", 1L))

// This is a def so that it works during checkpoint recovery:
private def clock = ssc.scheduler.clock
Expand Down