[SPARK-30362][Core] Update InputMetrics in DataSourceRDD #27021

sandeep-katta · 2019-12-27T03:56:12Z

What changes were proposed in this pull request?

Incase of DS v2 InputMetrics are not updated

Before Fix

After Fix we can see that Input Size / Records is updated in the UI

Why are the changes needed?

InputMetrics like bytesread and recordread should be updated

Does this PR introduce any user-facing change?

No

How was this patch tested?

Added UT and also verified manually

sandeep-katta · 2019-12-27T04:18:47Z

@HyukjinKwon @cloud-fan

cloud-fan · 2019-12-27T05:56:35Z

sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/DataSourceRDD.scala

@@ -56,6 +58,15 @@ class DataSourceRDD(
    context.addTaskCompletionListener[Unit](_ => reader.close())
    val iter = new Iterator[Any] {
      private[this] var valuePrepared = false
+      private val inputMetrics = context.taskMetrics().inputMetrics
+      private val existingBytesRead = inputMetrics.bytesRead


I'm not sure if we can support this if the data source don't report the size metrics. AFAIK ds v1 doesn't support it either. We only support it in file source.

I think we can only support the "recordsRead" metrics for now. We need to design a general API for data sources to report metrics. cc @rdblue @brkyvz

ping @rdblue @brkyvz

+1 @cloud-fan's advice.

if we update only "recordsRead". In the UI "Input Size/Records" column will look like "0/ 10". Is this okay ?

that's true for DS v1 too, right?

yes it's true for DS V1 also. Shall we have some marker trait which says Input Size metrics is supported. For e.g. FilePartitionReaderFactory supports Input Size metrics. So that we can collect input size metrics only if it supports .

yes we should, but then it's an API design and we should involve more people to discuss.

then shall I update the solution and push, then people can comment on the approach ?

I'd suggest we send an email to dev list with your proposal to trigger discussion first. AFAIK many people are interested in the metrics API in DS v2.

SparkQA · 2019-12-27T08:00:46Z

Test build #115839 has finished for PR 27021 at commit 870ba7d.

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

sandeep-katta · 2020-01-27T06:39:31Z

@rdblue please review this, I have tested the changes from my end.

SparkQA · 2020-01-27T06:56:22Z

Test build #117431 has finished for PR 27021 at commit 1be7a33.

This patch fails to build.
This patch merges cleanly.
This patch adds the following public classes (experimental):
class PartitionIterator[T](reader: PartitionReader[T]) extends Iterator[T]
class MetricsHandler extends Logging with Serializable

SparkQA · 2020-01-27T07:19:36Z

Test build #117433 has finished for PR 27021 at commit 6c87a41.

This patch fails to build.
This patch merges cleanly.
This patch adds the following public classes (experimental):
class PartitionIterator[T](reader: PartitionReader[T]) extends Iterator[T]
class MetricsHandler extends Logging with Serializable

SparkQA · 2020-01-27T09:20:45Z

Test build #117437 has finished for PR 27021 at commit df4bff9.

This patch fails to build.
This patch merges cleanly.
This patch adds the following public classes (experimental):
class PartitionIterator[T](reader: PartitionReader[T]) extends Iterator[T]
class MetricsHandler extends Logging with Serializable

sandeep-katta · 2020-01-27T12:35:36Z

this is weird :( , I am able to build successfully using mvn but sbt is failing .

SparkQA · 2020-01-27T18:45:30Z

Test build #117445 has finished for PR 27021 at commit 5bd4b7b.

This patch passes all tests.
This patch merges cleanly.
This patch adds the following public classes (experimental):
class PartitionIterator[T](reader: PartitionReader[T]) extends Iterator[T]
class MetricsHandler extends Logging with Serializable

rdblue · 2020-01-27T21:18:34Z

sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/DataSourceRDD.scala

+  }
+}
+
+class MetricsHandler extends Logging with Serializable {


Should these new classes be private instead? I don't see a need to expose them.

rdblue · 2020-01-27T21:20:50Z

+1 for this change, with a fix to avoid exposing the new helper classes.

To address @cloud-fan's objection, this solution records the amount of data read by Hadoop file systems. We can always expose an additional way for v2 sources to return a size metric if the bytes read by those sources do not go through the Hadoop FileSystem API, but there are many cases that do use the file system API and at least those are supported by this change.

Thanks @sandeep-katta!

SparkQA · 2020-01-28T08:05:02Z

Test build #117465 has finished for PR 27021 at commit 532fd1f.

This patch fails due to an unknown error code, -9.
This patch merges cleanly.
This patch adds no public classes.

sandeep-katta · 2020-01-28T09:12:34Z

retest this please

SparkQA · 2020-01-28T13:06:18Z

Test build #117478 has finished for PR 27021 at commit 532fd1f.

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

cloud-fan · 2020-01-30T13:11:51Z

sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/DataSourceRDD.scala

+  }
+}
+
+private class MetricsBatchIterator(


nit: to be consistent, let's have a MetricsRowIterator, and the base class MetricsIterator doesn't need to implement next

or the base class can implement next as

protected def numRecords(element: T): Int override def next(): I = { val item = iter.next metricsHandler.updateMetrics(numRecords(item)) item }

cloud-fan · 2020-01-30T13:15:05Z

sql/core/src/test/scala/org/apache/spark/sql/execution/DataSourceScanExecRedactionSuite.scala

@@ -167,4 +170,31 @@ class DataSourceV2ScanExecRedactionSuite extends DataSourceScanRedactionTest {
      }
    }
  }
+
+  test("SPARK-30362: test input metrics for DSV2") {


to actually test DSV2, let's call withSQLConf in the test and set SQLConf.USE_V1_SOURCE_LIST to empty.

cloud-fan · 2020-01-30T13:16:25Z

OK let's get in this workaround, but we do need to think of a metrics API for DS v2 later.

LGTM except a few minor comments.

SparkQA · 2020-01-30T18:06:46Z

Test build #117569 has finished for PR 27021 at commit 34425ad.

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

cloud-fan · 2020-01-31T06:01:43Z

thanks, merging to master!

Incase of DS v2 InputMetrics are not updated **Before Fix** ![inputMetrics](https://user-images.githubusercontent.com/35216143/71501010-c216df00-288d-11ea-8522-fdd50b13eae1.png) **After Fix** we can see that `Input Size / Records` is updated in the UI ![image](https://user-images.githubusercontent.com/35216143/71501000-b88d7700-288d-11ea-92fe-a727b2b79908.png) InputMetrics like bytesread and recordread should be updated No Added UT and also verified manually Closes apache#27021 from sandeep-katta/dsv2inputmetrics. Authored-by: sandeep katta <sandeep.katta2007@gmail.com> Signed-off-by: Wenchen Fan <wenchen@databricks.com> ---- Cherry-pick conflicts: - sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/DataSourceRDD.scala only import statements, as upstream reorganized packages - sql/core/src/test/scala/org/apache/spark/sql/execution/DataSourceScanExecRedactionSuite.scala The test in original PR sets a SQLConf to empty string that doesnt exist yet. Will need to be reintroduced after pulling from master.

* [SPARK-30362][CORE] Update InputMetrics in DataSourceRDD Incase of DS v2 InputMetrics are not updated **Before Fix** ![inputMetrics](https://user-images.githubusercontent.com/35216143/71501010-c216df00-288d-11ea-8522-fdd50b13eae1.png) **After Fix** we can see that `Input Size / Records` is updated in the UI ![image](https://user-images.githubusercontent.com/35216143/71501000-b88d7700-288d-11ea-92fe-a727b2b79908.png) InputMetrics like bytesread and recordread should be updated No Added UT and also verified manually Closes apache#27021 from sandeep-katta/dsv2inputmetrics. Authored-by: sandeep katta <sandeep.katta2007@gmail.com> Signed-off-by: Wenchen Fan <wenchen@databricks.com> ---- Cherry-pick conflicts: - sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/DataSourceRDD.scala only import statements, as upstream reorganized packages - sql/core/src/test/scala/org/apache/spark/sql/execution/DataSourceScanExecRedactionSuite.scala The test in original PR sets a SQLConf to empty string that doesnt exist yet. Will need to be reintroduced after pulling from master. * waitUntilEmpty requires a timeout until 5e92301 Co-authored-by: sandeep katta <sandeep.katta2007@gmail.com>

Incase of DS v2 InputMetrics are not updated **Before Fix** ![inputMetrics](https://user-images.githubusercontent.com/35216143/71501010-c216df00-288d-11ea-8522-fdd50b13eae1.png) **After Fix** we can see that `Input Size / Records` is updated in the UI ![image](https://user-images.githubusercontent.com/35216143/71501000-b88d7700-288d-11ea-92fe-a727b2b79908.png) InputMetrics like bytesread and recordread should be updated No Added UT and also verified manually Closes apache#27021 from sandeep-katta/dsv2inputmetrics. Authored-by: sandeep katta <sandeep.katta2007@gmail.com> Signed-off-by: Wenchen Fan <wenchen@databricks.com> RB=2618349 BUG=LIHADOOP-57737 G=spark-reviewers R=vsowrira,mshen A=vsowrira

cloud-fan reviewed Dec 27, 2019

View reviewed changes

sandeep-katta force-pushed the dsv2inputmetrics branch from 870ba7d to 1be7a33 Compare January 27, 2020 06:38

sandeep-katta force-pushed the dsv2inputmetrics branch from 1be7a33 to 6c87a41 Compare January 27, 2020 07:07

sandeep-katta force-pushed the dsv2inputmetrics branch from 6c87a41 to df4bff9 Compare January 27, 2020 09:01

Update inputmetrics in DataSourceRDD

5bd4b7b

sandeep-katta force-pushed the dsv2inputmetrics branch from df4bff9 to 5bd4b7b Compare January 27, 2020 14:35

rdblue reviewed Jan 27, 2020

View reviewed changes

fixed review comments

532fd1f

cloud-fan reviewed Jan 30, 2020

View reviewed changes

fixed review comments

34425ad

cloud-fan closed this in 5f3ec62 Jan 31, 2020

JacekLach mentioned this pull request May 13, 2020

[SPARK-30362][CORE] Update InputMetrics in DataSourceRDD palantir/spark#680

Merged

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

[SPARK-30362][Core] Update InputMetrics in DataSourceRDD #27021

[SPARK-30362][Core] Update InputMetrics in DataSourceRDD #27021

sandeep-katta commented Dec 27, 2019 •

edited

sandeep-katta commented Dec 27, 2019

cloud-fan Dec 27, 2019

sandeep-katta Jan 7, 2020

HyukjinKwon Jan 17, 2020

sandeep-katta Jan 17, 2020 •

edited

cloud-fan Jan 17, 2020

sandeep-katta Jan 17, 2020

cloud-fan Jan 17, 2020 •

edited

sandeep-katta Jan 17, 2020

cloud-fan Jan 17, 2020

SparkQA commented Dec 27, 2019

sandeep-katta commented Jan 27, 2020

SparkQA commented Jan 27, 2020

SparkQA commented Jan 27, 2020

SparkQA commented Jan 27, 2020

sandeep-katta commented Jan 27, 2020

SparkQA commented Jan 27, 2020

rdblue Jan 27, 2020

rdblue commented Jan 27, 2020

SparkQA commented Jan 28, 2020

sandeep-katta commented Jan 28, 2020

SparkQA commented Jan 28, 2020

cloud-fan Jan 30, 2020

cloud-fan Jan 30, 2020

cloud-fan Jan 30, 2020

cloud-fan commented Jan 30, 2020

SparkQA commented Jan 30, 2020

cloud-fan commented Jan 31, 2020

[SPARK-30362][Core] Update InputMetrics in DataSourceRDD #27021

[SPARK-30362][Core] Update InputMetrics in DataSourceRDD #27021

Conversation

sandeep-katta commented Dec 27, 2019 • edited

What changes were proposed in this pull request?

Why are the changes needed?

Does this PR introduce any user-facing change?

How was this patch tested?

sandeep-katta commented Dec 27, 2019

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

sandeep-katta Jan 17, 2020 • edited

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

cloud-fan Jan 17, 2020 • edited

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

SparkQA commented Dec 27, 2019

sandeep-katta commented Jan 27, 2020

SparkQA commented Jan 27, 2020

SparkQA commented Jan 27, 2020

SparkQA commented Jan 27, 2020

sandeep-katta commented Jan 27, 2020

SparkQA commented Jan 27, 2020

Choose a reason for hiding this comment

rdblue commented Jan 27, 2020

SparkQA commented Jan 28, 2020

sandeep-katta commented Jan 28, 2020

SparkQA commented Jan 28, 2020

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

cloud-fan commented Jan 30, 2020

SparkQA commented Jan 30, 2020

cloud-fan commented Jan 31, 2020

sandeep-katta commented Dec 27, 2019 •

edited

sandeep-katta Jan 17, 2020 •

edited

cloud-fan Jan 17, 2020 •

edited