[SPARK-26164][SQL] Allow FileFormatWriter to write multiple partitions/buckets without sort by c21 · Pull Request #23163 · apache/spark

c21 · 2018-11-28T10:01:48Z

What changes were proposed in this pull request?

Currently spark always requires a local sort before writing to output table on partition/bucket columns (see write.requiredOrdering in FileFormatWriter.scala), which is unnecessary, and can be avoided by keeping multiple output writers concurrently in FileFormatDataWriter.scala.

This pr is first doing hash-based write, then falling back to sort-based write (current implementation) when number of opened writer exceeding a threshold (controlled by a config). Specifically:

(hash-based write) Maintain mapping between file path and output writer, and re-use writer for writing input row. In case of the number of opened output writers exceeding a threshold (can be changed by a config), we go to 2.
(sort-based write) Sort the rest of input rows (use the same sorter in SortExec). Then writing the rest of sorted rows, and we can close the writer on the fly, in case no more rows for current file path.

How was this patch tested?

Added unit test in DataFrameReaderWriterSuite.scala. Existing test like SQLMetricsSuite.scala would already exercise the code path of executor write metrics.

c21 · 2018-11-28T10:07:45Z

cc people who have most context for review - @cloud-fan, @tejasapatil and @sameeragarwal. Thanks!

gatorsmile · 2018-11-28T22:57:38Z

add to whitelist

gatorsmile · 2018-11-28T23:00:47Z

@c21 Any perf number?

SparkQA · 2018-11-29T00:33:22Z

Test build #99412 has finished for PR 23163 at commit c2e81eb.

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

c21 · 2018-11-29T06:52:26Z

@gatorsmile:

Any perf number?

From my employer company workload for query writing dynamic partitions, we see >20% reserved CPU time (executor wall clock time) reduction, and >20% disk spill size reduction, after rolling out the change to use concurrent writers instead of sort (i.e. hash-based write in this pr).

I am not sure whether it's the performance number you were looking for. Let me know if anything needed. Thanks.

In addition, I updated the pr, as I found I need to change BasicWriteTaskStatsTracker as well.

SparkQA · 2018-11-29T08:05:02Z

Test build #99432 has finished for PR 23163 at commit a7ddb22.

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

c21 · 2018-12-01T16:46:42Z

cc @cloud-fan and @gatorsmile:

I think this pr is ready for review. Could you guys take a look when you have time? Thanks!
The test failure (fails due to an unknown error code, -9) seems to be unrelated to my change.

gatorsmile · 2018-12-02T19:22:37Z

retest this please

SparkQA · 2018-12-02T23:23:17Z

Test build #99579 has finished for PR 23163 at commit a7ddb22.

This patch fails Spark unit tests.
This patch does not merge cleanly.
This patch adds no public classes.

SparkQA · 2018-12-03T05:21:59Z

Test build #99585 has finished for PR 23163 at commit 6cb993b.

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

c21 · 2018-12-03T05:36:03Z

Jenkins, retest this please

SparkQA · 2018-12-03T07:10:09Z

Test build #99591 has finished for PR 23163 at commit 6cb993b.

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

c21 · 2018-12-03T20:10:38Z

Jenkins, retest this please

SparkQA · 2018-12-03T23:42:20Z

Test build #99624 has finished for PR 23163 at commit 6cb993b.

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

heary-cao · 2018-12-05T10:25:15Z

retest this please

SparkQA · 2018-12-05T14:50:23Z

Test build #99708 has finished for PR 23163 at commit 6cb993b.

This patch passes all tests.
This patch does not merge cleanly.
This patch adds no public classes.

SparkQA · 2019-01-03T08:05:02Z

Test build #100673 has finished for PR 23163 at commit 7c544ab.

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

c21 · 2019-01-03T16:57:26Z

Jenkins, retest this please

SparkQA · 2019-01-03T21:13:34Z

Test build #100696 has finished for PR 23163 at commit 7c544ab.

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

HyukjinKwon · 2019-01-04T02:30:11Z

retest this please

SparkQA · 2019-01-04T06:19:44Z

Test build #100714 has finished for PR 23163 at commit 7c544ab.

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

gatorsmile · 2019-06-06T21:07:33Z

cc @gengliangwang

AmplabJenkins · 2019-06-07T12:52:25Z

Can one of the admins verify this patch?

HyukjinKwon · 2019-09-17T00:25:00Z

ping @c21 to update or close

yizhu-wish · 2020-01-02T23:40:59Z

Is this still being actively worked on?

c21 · 2020-02-04T05:08:26Z

@HyukjinKwon, @yizhu-wish - I am resuming on this, and will have update in next few days, thanks.

github-actions · 2020-05-15T00:13:13Z

We're closing this PR because it hasn't been updated in a while. This isn't a judgement on the merit of the PR in any way. It's just a way of keeping the PR queue manageable.
If you'd like to revive this PR, please reopen it and ask a committer to remove the Stale tag!

…tions and bucket table ### What changes were proposed in this pull request? This is a re-proposal of #23163. Currently spark always requires a [local sort](https://github.com/apache/spark/blob/master/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/FileFormatWriter.scala#L188) before writing to output table with dynamic partition/bucket columns. The sort can be unnecessary if cardinality of partition/bucket values is small, and can be avoided by keeping multiple output writers concurrently. This PR introduces a config `spark.sql.maxConcurrentOutputFileWriters` (which disables this feature by default), where user can tune the maximal number of concurrent writers. The config is needed here as we cannot keep arbitrary number of writers in task memory which can cause OOM (especially for Parquet/ORC vectorization writer). The feature is to first use concurrent writers to write rows. If the number of writers exceeds the above config specified limit. Sort rest of rows and write rows one by one (See `DynamicPartitionDataConcurrentWriter.writeWithIterator()`). In addition, interface `WriteTaskStatsTracker` and its implementation `BasicWriteTaskStatsTracker` are also changed because previously they are relying on the assumption that only one writer is active for writing dynamic partitions and bucketed table. ### Why are the changes needed? Avoid the sort before writing output for dynamic partitioned query and bucketed table. Help improve CPU and IO performance for these queries. ### Does this PR introduce _any_ user-facing change? No. ### How was this patch tested? Added unit test in `DataFrameReaderWriterSuite.scala`. Closes#32198 from c21/writer. Authored-by: Cheng Su <chengsu@fb.com> Signed-off-by: Wenchen Fan <wenchen@databricks.com>

c21 force-pushed the more-writers branch from c2e81eb to a7ddb22Compare November 29, 2018 06:26

c21 force-pushed the more-writers branch from a7ddb22 to 6cb993bCompare December 3, 2018 02:06

c21 added 2 commits January 2, 2019 21:39

Fix the metrics logic in BasicWriteTaskStatsTracker accordingly
5f43f57

Fix the unit test in BasicWriteTaskStatsTrackerSuite
7c544ab

c21 force-pushed the more-writers branch from 6cb993b to 7c544abCompare January 3, 2019 06:01

dongjoon-hyun added the SQL label Jun 14, 2019

github-actionsBot added the Stale label May 15, 2020

github-actionsBot closed this May 17, 2020

c21 mentioned this pull request Apr 16, 2021
[SPARK-26164][SQL] Allow concurrent writers for writing dynamic partitions and bucket table #32198
Closed

Uh oh!

Conversation

c21 commented Nov 28, 2018

What changes were proposed in this pull request?

How was this patch tested?

Uh oh!

c21 commented Nov 28, 2018

Uh oh!

gatorsmile commented Nov 28, 2018

Uh oh!

gatorsmile commented Nov 28, 2018

Uh oh!

SparkQA commented Nov 29, 2018

Uh oh!

c21 commented Nov 29, 2018• edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

SparkQA commented Nov 29, 2018

Uh oh!

c21 commented Dec 1, 2018

Uh oh!

gatorsmile commented Dec 2, 2018

Uh oh!

SparkQA commented Dec 2, 2018

Uh oh!

SparkQA commented Dec 3, 2018

Uh oh!

c21 commented Dec 3, 2018

Uh oh!

SparkQA commented Dec 3, 2018

Uh oh!

c21 commented Dec 3, 2018

Uh oh!

SparkQA commented Dec 3, 2018

Uh oh!

heary-cao commented Dec 5, 2018

Uh oh!

SparkQA commented Dec 5, 2018

Uh oh!

SparkQA commented Jan 3, 2019

Uh oh!

c21 commented Jan 3, 2019

Uh oh!

SparkQA commented Jan 3, 2019

Uh oh!

HyukjinKwon commented Jan 4, 2019

Uh oh!

SparkQA commented Jan 4, 2019

Uh oh!

gatorsmile commented Jun 6, 2019

Uh oh!

AmplabJenkins commented Jun 7, 2019

Uh oh!

HyukjinKwon commented Sep 17, 2019

Uh oh!

yizhu-wish commented Jan 2, 2020

Uh oh!

c21 commented Feb 4, 2020

Uh oh!

github-actionsBot commented May 15, 2020

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

8 participants

c21 commented Nov 29, 2018•
edited
Loading