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

emres · 2015-04-09T10:07:53Z

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.

…ember duration, with a default value of 1 minute.

srowen · 2015-04-09T11:11:36Z

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

   */
-  private val MIN_REMEMBER_DURATION = Minutes(1)
+  private val minRememberDuration = new SparkConf().get("spark.streaming.minRememberDuration", "1")


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?

@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.

srowen · 2015-04-09T11:32:38Z

(You can just push to this PR rather than close it.)

… reduced number of fields. * Changed the property name to spark.streaming.minRememberDurationMin to reflect the unit of value (minutes). * Deleted the constant MIN_REMEMBER_DURATION, because now minRememberDurationMin is serving the same purpose.

emres · 2015-04-09T11:47:09Z

@srowen I've pushed to my branch SPARK-3276 again. Is it acceptable now?

srowen · 2015-04-09T11:51:39Z

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

   */
-  private val MIN_REMEMBER_DURATION = Minutes(1)
+  private val minRememberDurationMin = Minutes(new SparkConf()


This much looks better, though now I also see that we need a real SparkConf here somehow. I don't think this is a global, static property of the object anymore. I think this would have to live in the FileInputDStream class now?

@srowen I'm a little confused. Do you mean something like: "inside private[streaming] class FileInputDStream[K, V, F <: NewInputFormat[K,V]] class create a private val conf = new SparkConf() and then use that conf inside private[streaming] object FileInputDStream"?

Or something totally different?

IIUC I think what @srowen mentions is to put minRememberDurationMin related code into class FileInputDStream, rather than in the object. By putting this into the class FileInputDStream, you could use conf from StreamingContext, no need to create a new one again.

Based on the comment of @jerryshao I prepared another commit, moving the minRememberDurationMin from the companion object to the class. @srowen does it look better now? Anything else to be done?

jhu-chang · 2015-04-10T09:02:48Z

This approach looks like affecting all the file source in one project, may be have some cases that parts of the file stream needs to include some old files, but others do not. Is a parameter more suitable for this?

* Moved the minRememberDurationMin to the class so that it can use the existing Spark context * Refactored calculateNumBatchesToRemember to take minRememberDurationMin as a parameter

emres · 2015-04-13T09:38:56Z

Based on the comment of @jerryshao I prepared another commit, moving the minRememberDurationMin from the companion object to the class. @srowen does it look better now? Anything else to be done?

jerryshao · 2015-04-13T09:53:08Z

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

+   */
+  private val minRememberDurationMin = Minutes(ssc.sparkContext.getConf
+                                               .get("spark.streaming.minRememberDurationMin", "1")
+                                               .toLong)


I think you can use ssc.conf rather than ssc.sparkContext.getConf, also SparkConf has a API getLong which can be directly used.

…nf, and also getLong method directly.

emres · 2015-04-13T10:04:42Z

Based on the comment of @jerryshao , the code is now using ssc.conf rather than ssc.sparkContext.getConf, and also getLong method directly.

srowen · 2015-04-13T12:11:55Z

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

+   * 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.conf


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.

emres · 2015-04-13T12:37:31Z

@srowen fair enough. It looks like if SPARK-5931 is resolved, I can go back to using spark.streaming.minRememberDuration as a Spark configuration property name, right? And I also think I will have to pass the value as "60s" (will there be a unit designation for 'minute's, or is it going to be only 's' for seconds, and 'ms' for milliseconds?)

Do you foresee any other changes? E.g., will there be a new method to retrieve the value of the duration parameter such as spark.streaming.minRememberDuration?

srowen · 2015-04-13T12:40:19Z

Yes, I think you can change the prop name. I don't see a "minutes" designator, but might as well use seconds. I can sort of imagine wanting to specify a value less than 60 seconds, so the extra granularity might be beneficial. I don't think there are any special methods being added for specific properties.

…rDuration * switched back to using spark.streaming.minRememberDuration * renamed minRememberDurationMin to minRememberDuration

emres · 2015-04-13T12:57:31Z

@srowen I've switched back to

spark.streaming.minRememberDuration as the property name (with a default value of 60 seconds)
minRememberDuration as the corresponding variable name

if these final changes are OK, then I'll be waiting for SPARK-5931 to be resolved.

emres · 2015-04-14T14:04:05Z

@srowen I see that SPARK-5931 is marked as CLOSED (Fixed), and the associated Pull Request has been closed. How does that affect this pull request of mine? Anything else to be done?

srowen · 2015-04-14T14:19:49Z

Yes, have a look at the changes. You can and should now create a property with a value like "60s" like the new properties in that PR.

* Switched to using newly added getTimeAsSeconds method (see apache#5236) * Renamed minRememberDuration to minRememberDurationS to be compatible with the examples in the pull request above.

emres · 2015-04-14T15:36:23Z

@srowen I prepared another commit in which I'm now using the newly added getTimeAsSeconds method and passing the default value as "60s". Anything else to be done?

srowen · 2015-04-14T15:47:30Z

That looks right. Still probably needs a nod from @tdas as it kinda adds a new API (config param)

emres · 2015-04-15T14:13:10Z

@srowen thanks for checking. @tdas does it look OK to you, too?

One thing I've realized is that I did not touch streaming programming guide, and I think I should document this. But after checking https://spark.apache.org/docs/latest/streaming-programming-guide.html, I could not be sure where to document this property. Where is the most appropriate place for documenting spark.streaming.minRememberDuration? Should I put it in the table given at https://spark.apache.org/docs/latest/configuration.html#spark-streaming?

srowen · 2015-04-19T11:31:09Z

I suspect this is OK. I wonder if we should simply leave this undocumented for now as it's somewhat obscure, rather than completely commit to it as an API. That might make this easier to merge.

emres · 2015-04-20T12:00:47Z

@srowen fair enough. I wonder if @tdas will have time to check this PR. (We are already planning to use this feature for one of our existing customers, it'd be great if it became part of the official master branch.)

srowen · 2015-04-21T20:40:17Z

ok to test

srowen · 2015-04-21T20:42:12Z

Oh, darn it. I merged this since I though this one had passed tests, but on looking back at this one, it has not yet. I will keep a close eye on this and make sure it actually does, and if not, will revert it right away.

srowen · 2015-04-21T20:45:24Z

Jenkins, test this please

srowen · 2015-04-22T00:14:43Z

So, I think I got away with this one. I tested it locally and it passes, and I see that subsequent pull requests in Jenkins succeed, and ones that failed did so for unrelated reasons. So this does actually pass tests. Of course it was an unintentional mistake to proceed before seeing the Jenkins result on this one explicitly, but no actual harm done in this case.

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 apache#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.

tdas · 2015-07-15T03:40:24Z

@srowen Another reason why this was premature merge was the configuration spark.streaming.minRememberDuration. Its so confusing. Is it immediately obvious that this is related to files?

srowen · 2015-07-15T07:51:36Z

Maybe so -- worth changing now?
How can I get your attention to review though? this was months ago. I pinged you above twice.

tdas · 2015-07-15T08:52:17Z

Yes, I am a partial culprit in this matter, as I was not able to give the
necessary attention. So the only way forward is to ensure that I can
communicate more of my review knowledge to you. :) Consider this as a
nugget, configuration names should make it very evident what it is related
to.

On Wed, Jul 15, 2015 at 12:52 AM, Sean Owen [email protected]
wrote:

Maybe so -- worth changing now?
How can I get your attention to review though? this was months ago. I
pinged you above twice.

—
Reply to this email directly or view it on GitHub
#5438 (comment).

SPARK-3276 Added a new configuration parameter spark.streaming.minRem…

43cc1ce

…ember duration, with a default value of 1 minute.

srowen reviewed Apr 9, 2015
View reviewed changes

SPARK-3276 Moved the minRememberDurationMin to the class

bfe0acb

* Moved the minRememberDurationMin to the class so that it can use the existing Spark context * Refactored calculateNumBatchesToRemember to take minRememberDurationMin as a parameter

jerryshao reviewed Apr 13, 2015
View reviewed changes

SPARK-3276 Started to use ssc.conf rather than ssc.sparkContext.getCo…

1c53ba9

…nf, and also getLong method directly.

srowen reviewed Apr 13, 2015
View reviewed changes

SPARK-3276 Minor code re-formatting.

c9d58ca

SPARK-3276 Changed the property name and variable name for minRemembe…

affee1d

…rDuration * switched back to using spark.streaming.minRememberDuration * renamed minRememberDurationMin to minRememberDuration

SPARK-3276 Switched to using newly added getTimeAsSeconds method.

766f938

* Switched to using newly added getTimeAsSeconds method (see apache#5236) * Renamed minRememberDuration to minRememberDurationS to be compatible with the examples in the pull request above.

asfgit closed this in c25ca7c Apr 21, 2015

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

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

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

emres commented Apr 9, 2015

srowen Apr 9, 2015

emres Apr 9, 2015

srowen commented Apr 9, 2015

emres commented Apr 9, 2015

srowen Apr 9, 2015

emres Apr 9, 2015

jerryshao Apr 10, 2015

emres Apr 13, 2015

jhu-chang commented Apr 10, 2015

emres commented Apr 13, 2015

jerryshao Apr 13, 2015

emres commented Apr 13, 2015

srowen Apr 13, 2015

emres commented Apr 13, 2015

srowen commented Apr 13, 2015

emres commented Apr 13, 2015

emres commented Apr 14, 2015

srowen commented Apr 14, 2015

emres commented Apr 14, 2015

srowen commented Apr 14, 2015

emres commented Apr 15, 2015

srowen commented Apr 19, 2015

emres commented Apr 20, 2015

srowen commented Apr 21, 2015

srowen commented Apr 21, 2015

srowen commented Apr 21, 2015

srowen commented Apr 22, 2015

tdas commented Jul 15, 2015

srowen commented Jul 15, 2015

tdas commented Jul 15, 2015

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

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

Conversation

emres commented Apr 9, 2015

Choose a reason for hiding this comment

Choose a reason for hiding this comment

srowen commented Apr 9, 2015

emres commented Apr 9, 2015

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

jhu-chang commented Apr 10, 2015

emres commented Apr 13, 2015

Choose a reason for hiding this comment

emres commented Apr 13, 2015

Choose a reason for hiding this comment

emres commented Apr 13, 2015

srowen commented Apr 13, 2015

emres commented Apr 13, 2015

emres commented Apr 14, 2015

srowen commented Apr 14, 2015

emres commented Apr 14, 2015

srowen commented Apr 14, 2015

emres commented Apr 15, 2015

srowen commented Apr 19, 2015

emres commented Apr 20, 2015

srowen commented Apr 21, 2015

srowen commented Apr 21, 2015

srowen commented Apr 21, 2015

srowen commented Apr 22, 2015

tdas commented Jul 15, 2015

srowen commented Jul 15, 2015

tdas commented Jul 15, 2015