[SPARK-26193][SQL] Implement shuffle write metrics in SQL

[xuanyuanking]

What changes were proposed in this pull request?

1. Implement SQLShuffleWriteMetricsReporter on the SQL side as the customized ShuffleWriteMetricsReporter.
2. Add shuffle write metrics to ShuffleExchangeExec, and use these metrics to create corresponding SQLShuffleWriteMetricsReporter in shuffle dependency.
3. Rework on ShuffleMapTask to add new class named ShuffleWriteProcessor which control shuffle write process, we use sql shuffle write metrics by customizing a ShuffleWriteProcessor on SQL side.

How was this patch tested?

Add UT in SQLMetricsSuite.
Manually test locally, update screen shot to document attached in JIRA.

[xuanyuanking]

test this please.

[SparkQA]

Test build #99616 has finished for PR 23207 at commit 7c8e516.

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

[SparkQA]

Test build #99617 has finished for PR 23207 at commit 7c8e516.

• This patch fails from timeout after a configured wait of 400m.
• This patch merges cleanly.
• This patch adds no public classes.

[xuanyuanking]

retest this please.

[SparkQA]

Test build #99643 has finished for PR 23207 at commit 7c8e516.

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

[xuanyuanking]

cc @cloud-fan @gatorsmile @rxin

[cloud-fan]

Can you share some ideas about it? IMO shuffle write metrics is hard, as an RDD can have shuffle dependencies with multiple upstream RDDs. That said, in general the shuffle write metrics should belong to the upstream RDDs.

In Spark SQL, it's a little simpler, as the ShuffledRowRDD always have only one child, so it's reasonable to say that shuffle write metrics belong to ShuffledRowRDD.

That said, we need to design a not-so-general shuffle write metrics API in Spark core, which will only be used in Spark SQL.

[xuanyuanking]

Thanks for your reply Wenchen, there's a sketch doc assigned in JIRA:https://docs.google.com/document/d/1DX0gLkpk_NCE5MwI1_m4gnA2rLdjDkynZ02u2VWDR-8/edit

IMO shuffle write metrics is hard, as an RDD can have shuffle dependencies with multiple upstream RDDs. That said, in general the shuffle write metrics should belong to the upstream RDDs.

That's right and that's also what I try to do at first, logically upstream operator trigger shuffle write, and first attempt implementation is also changed SparkPlan base class to achieve this.

In Spark SQL, it's a little simpler, as the ShuffledRowRDD always have only one child, so it's reasonable to say that shuffle write metrics belong to ShuffledRowRDD.

Yes, maybe this also the suggestion by Reynold, ShuffleExchangeExec has only one child, we can do a simplify on the implementation. But as the shuffle write metrics are updated by task inner, so the core module still need some changes.

[SparkQA]

Test build #99677 has finished for PR 23207 at commit fcd62b3.

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

[SparkQA]

Test build #99676 has finished for PR 23207 at commit ca6c407.

• This patch fails Spark unit tests.
• This patch merges cleanly.
• This patch adds no public classes.

[SparkQA]

Test build #99722 has finished for PR 23207 at commit cf35b9f.

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

[SparkQA]

Test build #99736 has finished for PR 23207 at commit 76d1ca0.

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

[rxin]

@xuanyuanking can you separate the prs to rename read side metric and the write side change?

[xuanyuanking]

Can we put the above in a closure and pass it into shuffle dependency? Then in SQL we just put the above in SQL using custom metrics.

Yea, the commit of a780b70 achieve this by adding ShuffleWriteProcessor abstract.
And the read metrics rename reverted in 7d104eb, will do it and display change in another pr.

[SparkQA]

Test build #99782 has finished for PR 23207 at commit d5ee249.

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

[cloud-fan]

the code looks much cleaner now!

[xuanyuanking]

the code looks much cleaner now!

Sorry for the original rush and code, I should and will pay more attention on coding clean and more discussion on optional implementation.

[SparkQA]

Test build #99805 has finished for PR 23207 at commit 6378a3d.

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

[xuanyuanking]

retest this please.

[cloud-fan]

LGTM except a few comments. cc @rxin to take another look

[SparkQA]

Test build #99817 has finished for PR 23207 at commit 6378a3d.

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

[SparkQA]

Test build #99825 has finished for PR 23207 at commit bc2c4f1.

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

[cloud-fan]

thanks, merging to master!

[xuanyuanking]

Thanks all for your review.