[SPARK-4537][Streaming] Expand StreamingSource to add more metrics #3466

jerryshao · 2014-11-26T03:14:47Z

Add processingDelay, schedulingDelay and totalDelay for the last completed batch. Add lastReceivedBatchRecords and totalReceivedBatchRecords to the received records counting.

SparkQA · 2014-11-26T03:20:22Z

Test build #23868 has started for PR 3466 at commit c7a9376.

This patch merges cleanly.

SparkQA · 2014-11-26T04:48:39Z

Test build #23868 has finished for PR 3466 at commit c7a9376.

This patch passes all tests.
This patch merges cleanly.
This patch adds no public classes.

AmplabJenkins · 2014-11-26T04:48:42Z

Test PASSed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/23868/
Test PASSed.

maasg · 2014-11-26T09:38:26Z

streaming/src/main/scala/org/apache/spark/streaming/StreamingSource.scala

+  // Gauge for last received batch records and total received batch records.
+  private var totalReceivedBatchRecords: Long = 0L
+  def getTotalReceivedBatchRecords(listener: StreamingJobProgressListener): Long = {
+    totalReceivedBatchRecords += listener.lastReceivedBatchRecords.values.sum


Will this counter work? I think that gauges are collected only on request of a source, so if nobody is consuming the metric or consuming it too often, we will have a wrong count.

Aha, you're right, so it is hard to collect the total batch records without modifying the StreamingJobProgressListener code.

I will fix this, thanks a lot.

SparkQA · 2014-11-27T05:55:08Z

Test build #23913 has started for PR 3466 at commit 02dd44f.

This patch merges cleanly.

SparkQA · 2014-11-27T07:15:17Z

Test build #23913 has finished for PR 3466 at commit 02dd44f.

This patch passes all tests.
This patch merges cleanly.
This patch adds no public classes.

AmplabJenkins · 2014-11-27T07:15:21Z

Test PASSed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/23913/
Test PASSed.

tdas · 2014-12-23T23:50:30Z

streaming/src/main/scala/org/apache/spark/streaming/StreamingSource.scala

@@ -62,6 +62,14 @@ private[streaming] class StreamingSource(ssc: StreamingContext) extends Source {
  registerGauge("lastCompletedBatch_processEndTime",
    _.lastCompletedBatch.flatMap(_.processingEndTime).getOrElse(-1L), -1L)

+  // Gauge for last completed batch's delay information.
+  registerGauge("lastCompletedBatch_processTime",


Its better to name this "processingTime"

jerryshao · 2014-12-24T00:55:03Z

Thanks a lot TD for your comments, I will factor out the old code and fix the above issues you mentioned.

SparkQA · 2014-12-24T01:47:41Z

Test build #24751 has started for PR 3466 at commit c097ddc.

This patch merges cleanly.

jerryshao · 2014-12-24T01:48:13Z

Hey TD, I've addressed the problem you mentioned in this way, I'm not is this what you want, would you mind taking a look at it ? Thanks a lot.

tdas · 2014-12-24T01:59:32Z

streaming/src/main/scala/org/apache/spark/streaming/StreamingSource.scala

      defaultValue: T) {
    metricRegistry.register(MetricRegistry.name("streaming", name), new Gauge[T] {
-      override def getValue: T = Option(f(streamingListener)).getOrElse(defaultValue)


I think its better to keep the Option here (and document that defaultValue is used when f returns null. And other places should not have to use Option. This is safer for any one to use and also minimizes the changes.

OK, I will revert it back and try a better way.

Hi TD, what's your meaning of "And other places should not have to use Option", If here as an example, change to

registerGauge("lastCompletedBatch_submissionTime", _.lastCompletedBatch.map(_.submissionTime).get, -1L)

get will throw exception when there's no completed batch. I'm not sure what's actual meaning, sorry if I misunderstand anything.

Hi TD, sorry to bother you again, I'm not if there's a better way to address this problem, would you mind giving me some hints, thanks a lot.

I understand the problem. Good catch, I did not realize that. How about this. Lets make two versions of registerGauge, one that takes f: StreamingProgressListener => T without any default value, another that takes f: StreamingProgressListener => Option[T] and the default value. Each version will be used accordingly.

OK, got it, I will change the code as you suggested.

SparkQA · 2014-12-24T03:36:35Z

Test build #24751 has finished for PR 3466 at commit c097ddc.

This patch passes all tests.
This patch merges cleanly.
This patch adds no public classes.

AmplabJenkins · 2014-12-24T03:36:39Z

Test PASSed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/24751/
Test PASSed.

SparkQA · 2014-12-25T05:22:28Z

Test build #24808 has started for PR 3466 at commit 44721a6.

This patch merges cleanly.

SparkQA · 2014-12-25T06:47:49Z

Test build #24808 has finished for PR 3466 at commit 44721a6.

This patch passes all tests.
This patch merges cleanly.
This patch adds no public classes.

AmplabJenkins · 2014-12-25T06:47:53Z

Test PASSed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/24808/
Test PASSed.

tdas · 2014-12-25T23:43:36Z

streaming/src/main/scala/org/apache/spark/streaming/StreamingSource.scala

@@ -35,6 +35,15 @@ private[streaming] class StreamingSource(ssc: StreamingContext) extends Source {
    })


nit: The existing version of registerGauge could have used the new version. Not a big deal, very small amount of duplicate code.

tdas · 2014-12-25T23:54:57Z

Just a couple of more comments for making the name more consistent with existing ones. Otherwise I approve of the how the registerGauge works now.

SparkQA · 2014-12-26T02:27:34Z

Test build #24827 has started for PR 3466 at commit 00f5f7f.

This patch merges cleanly.

jerryshao · 2014-12-26T02:27:47Z

Hi TD, thanks a lot for your comments, I just change the code style as you suggested, also add one more metrics totalProcessedRecords, would you mind reviewing this again? Thanks a lot and appreciate your time.

tdas · 2014-12-26T03:39:21Z

LGTM. Merging this. Thanks @jerryshao

Add `processingDelay`, `schedulingDelay` and `totalDelay` for the last completed batch. Add `lastReceivedBatchRecords` and `totalReceivedBatchRecords` to the received records counting. Author: jerryshao <saisai.shao@intel.com> Closes #3466 from jerryshao/SPARK-4537 and squashes the following commits: 00f5f7f [jerryshao] Change the code style and add totalProcessedRecords 44721a6 [jerryshao] Further address the comments c097ddc [jerryshao] Address the comments 02dd44f [jerryshao] Fix the addressed comments c7a9376 [jerryshao] Expand StreamingSource to add more metrics (cherry picked from commit f205fe4) Signed-off-by: Tathagata Das <tathagata.das1565@gmail.com>

SparkQA · 2014-12-26T03:52:02Z

Test build #24827 has finished for PR 3466 at commit 00f5f7f.

This patch passes all tests.
This patch merges cleanly.
This patch adds no public classes.

AmplabJenkins · 2014-12-26T03:52:05Z

Test PASSed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/24827/
Test PASSed.

Expand StreamingSource to add more metrics

c7a9376

maasg reviewed Nov 26, 2014
View reviewed changes

Fix the addressed comments

02dd44f

tdas reviewed Dec 23, 2014
View reviewed changes

Address the comments

c097ddc

tdas reviewed Dec 24, 2014
View reviewed changes

Further address the comments

44721a6

tdas reviewed Dec 25, 2014
View reviewed changes

Change the code style and add totalProcessedRecords

00f5f7f

asfgit closed this in f205fe4 Dec 26, 2014

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

[SPARK-4537][Streaming] Expand StreamingSource to add more metrics #3466

[SPARK-4537][Streaming] Expand StreamingSource to add more metrics #3466

jerryshao commented Nov 26, 2014

SparkQA commented Nov 26, 2014

SparkQA commented Nov 26, 2014

AmplabJenkins commented Nov 26, 2014

maasg Nov 26, 2014

jerryshao Nov 26, 2014

jerryshao Nov 26, 2014

SparkQA commented Nov 27, 2014

SparkQA commented Nov 27, 2014

AmplabJenkins commented Nov 27, 2014

tdas Dec 23, 2014

jerryshao commented Dec 24, 2014

SparkQA commented Dec 24, 2014

jerryshao commented Dec 24, 2014

tdas Dec 24, 2014

jerryshao Dec 24, 2014

jerryshao Dec 24, 2014

jerryshao Dec 24, 2014

tdas Dec 25, 2014

jerryshao Dec 25, 2014

SparkQA commented Dec 24, 2014

AmplabJenkins commented Dec 24, 2014

SparkQA commented Dec 25, 2014

SparkQA commented Dec 25, 2014

AmplabJenkins commented Dec 25, 2014

tdas Dec 25, 2014

tdas commented Dec 25, 2014

SparkQA commented Dec 26, 2014

jerryshao commented Dec 26, 2014

tdas commented Dec 26, 2014

SparkQA commented Dec 26, 2014

AmplabJenkins commented Dec 26, 2014

		@@ -35,6 +35,15 @@ private[streaming] class StreamingSource(ssc: StreamingContext) extends Source {
		})

[SPARK-4537][Streaming] Expand StreamingSource to add more metrics #3466

[SPARK-4537][Streaming] Expand StreamingSource to add more metrics #3466

Conversation

jerryshao commented Nov 26, 2014

SparkQA commented Nov 26, 2014

SparkQA commented Nov 26, 2014

AmplabJenkins commented Nov 26, 2014

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

SparkQA commented Nov 27, 2014

SparkQA commented Nov 27, 2014

AmplabJenkins commented Nov 27, 2014

Choose a reason for hiding this comment

jerryshao commented Dec 24, 2014

SparkQA commented Dec 24, 2014

jerryshao commented Dec 24, 2014

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

Choose a reason for hiding this comment

Choose a reason for hiding this comment

SparkQA commented Dec 24, 2014

AmplabJenkins commented Dec 24, 2014

SparkQA commented Dec 25, 2014

SparkQA commented Dec 25, 2014

AmplabJenkins commented Dec 25, 2014

Choose a reason for hiding this comment

tdas commented Dec 25, 2014

SparkQA commented Dec 26, 2014

jerryshao commented Dec 26, 2014

tdas commented Dec 26, 2014

SparkQA commented Dec 26, 2014

AmplabJenkins commented Dec 26, 2014