[SPARK-8890][SQL] Fallback on sorting when writing many dynamic partitions #8010

marmbrus · 2015-08-06T22:22:25Z

Previously, we would open a new file for each new dynamic written out using HadoopFsRelation. For formats like parquet this is very costly due to the buffers required to get good compression. In this PR I refactor the code allowing us to fall back on an external sort when many partitions are seen. As such each task will open no more than spark.sql.sources.maxFiles files. I also did the following cleanup:

Instead of keying the file HashMap on an expensive to compute string representation of the partition, we now use a fairly cheap UnsafeProjection that avoids heap allocations.
The control flow for instantiating and invoking a writer container has been simplified. Now instead of switching in two places based on the use of partitioning, the specific writer container must implement a single method writeRows that is invoked using runJob.
InternalOutputWriter has been removed. Instead we have a private[sql] method writeInternal that converts and calls the public method. This method can be overridden by internal datasources to avoid the conversion. This change remove a lot of code duplication and per-row asInstanceOf checks.
commands.scala has been split up.

…tions

SparkQA · 2015-08-06T22:30:08Z

Test build #40089 has finished for PR 8010 at commit 71cc717.

This patch fails RAT tests.
This patch merges cleanly.
This patch adds the following public classes (experimental):
- s"Using output committer class $
- logInfo(s"Using user defined output committer class $
- s"Using output committer class $

davies · 2015-08-06T22:53:05Z

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

If there is an exception happen before this line, currentWriter will not be closed.

SparkQA · 2015-08-06T23:54:23Z

Test build #40104 has finished for PR 8010 at commit 7e2d0a4.

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

SparkQA · 2015-08-07T02:53:57Z

Test build #40117 has finished for PR 8010 at commit f5675bd.

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

rxin · 2015-08-07T04:22:26Z

sql/core/src/main/scala/org/apache/spark/sql/SQLConf.scala

sources.maxConcurrentWrites ?

liancheng · 2015-08-07T12:17:31Z

LGTM except for some minor issues.

SparkQA · 2015-08-07T20:45:25Z

Test build #40183 has finished for PR 8010 at commit 40f0372.

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

SparkQA · 2015-08-07T21:06:18Z

Test build #40189 has finished for PR 8010 at commit 17b690e.

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

SparkQA · 2015-08-07T23:23:23Z

Test build #40193 has finished for PR 8010 at commit 00804fe.

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

…itions Previously, we would open a new file for each new dynamic written out using `HadoopFsRelation`. For formats like parquet this is very costly due to the buffers required to get good compression. In this PR I refactor the code allowing us to fall back on an external sort when many partitions are seen. As such each task will open no more than `spark.sql.sources.maxFiles` files. I also did the following cleanup: - Instead of keying the file HashMap on an expensive to compute string representation of the partition, we now use a fairly cheap UnsafeProjection that avoids heap allocations. - The control flow for instantiating and invoking a writer container has been simplified. Now instead of switching in two places based on the use of partitioning, the specific writer container must implement a single method `writeRows` that is invoked using `runJob`. - `InternalOutputWriter` has been removed. Instead we have a `private[sql]` method `writeInternal` that converts and calls the public method. This method can be overridden by internal datasources to avoid the conversion. This change remove a lot of code duplication and per-row `asInstanceOf` checks. - `commands.scala` has been split up. Author: Michael Armbrust <[email protected]> Closes #8010 from marmbrus/fsWriting and squashes the following commits: 00804fe [Michael Armbrust] use shuffleMemoryManager.pageSizeBytes 775cc49 [Michael Armbrust] Merge remote-tracking branch 'origin/master' into fsWriting 17b690e [Michael Armbrust] remove comment 40f0372 [Michael Armbrust] address comments f5675bd [Michael Armbrust] char -> string 7e2d0a4 [Michael Armbrust] make sure we close current writer 8100100 [Michael Armbrust] delete empty commands.scala 71cc717 [Michael Armbrust] update comment 8ec75ac [Michael Armbrust] [SPARK-8890][SQL] Fallback on sorting when writing many dynamic partitions (cherry picked from commit 49702bd) Signed-off-by: Michael Armbrust <[email protected]>

marmbrus · 2015-08-07T23:25:40Z

Thanks for reviewing. Merged to master and 1.5

yjshen · 2015-08-12T15:05:22Z

...e/src/main/scala/org/apache/spark/sql/execution/datasources/InsertIntoHadoopFsRelation.scala

Seems partitionColumnsInSpec and partitionColumns point to the same thing?
I think you mean the partition in InsertIntoTable? If so, it's already checked by PreWriteCheck.

Hmm, thats possible. I just copied this from the earlier code and moved it to a better place. If you want to add an analysis test to make sure this error works and then can still remove this code that would be great!

…itions Previously, we would open a new file for each new dynamic written out using `HadoopFsRelation`. For formats like parquet this is very costly due to the buffers required to get good compression. In this PR I refactor the code allowing us to fall back on an external sort when many partitions are seen. As such each task will open no more than `spark.sql.sources.maxFiles` files. I also did the following cleanup: - Instead of keying the file HashMap on an expensive to compute string representation of the partition, we now use a fairly cheap UnsafeProjection that avoids heap allocations. - The control flow for instantiating and invoking a writer container has been simplified. Now instead of switching in two places based on the use of partitioning, the specific writer container must implement a single method `writeRows` that is invoked using `runJob`. - `InternalOutputWriter` has been removed. Instead we have a `private[sql]` method `writeInternal` that converts and calls the public method. This method can be overridden by internal datasources to avoid the conversion. This change remove a lot of code duplication and per-row `asInstanceOf` checks. - `commands.scala` has been split up. Author: Michael Armbrust <[email protected]> Closes apache#8010 from marmbrus/fsWriting and squashes the following commits: 00804fe [Michael Armbrust] use shuffleMemoryManager.pageSizeBytes 775cc49 [Michael Armbrust] Merge remote-tracking branch 'origin/master' into fsWriting 17b690e [Michael Armbrust] remove comment 40f0372 [Michael Armbrust] address comments f5675bd [Michael Armbrust] char -> string 7e2d0a4 [Michael Armbrust] make sure we close current writer 8100100 [Michael Armbrust] delete empty commands.scala 71cc717 [Michael Armbrust] update comment 8ec75ac [Michael Armbrust] [SPARK-8890][SQL] Fallback on sorting when writing many dynamic partitions

marmbrus added 2 commits August 6, 2015 15:16

[SPARK-8890][SQL] Fallback on sorting when writing many dynamic parti…

8ec75ac

…tions

update comment

71cc717

marmbrus mentioned this pull request Aug 6, 2015

[SPARK-8890][SQL][WIP] Reduce memory consumption for dynamic partition insert #7514

Closed

delete empty commands.scala

8100100

davies reviewed Aug 6, 2015
View reviewed changes

make sure we close current writer

7e2d0a4

char -> string

f5675bd

rxin reviewed Aug 7, 2015
View reviewed changes

sql/core/src/main/scala/org/apache/spark/sql/SQLConf.scala Outdated

Copy link

Contributor

rxin Aug 7, 2015

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

sources.maxConcurrentWrites ?

marmbrus added 2 commits August 7, 2015 11:57

address comments

40f0372

remove comment

17b690e

Merge remote-tracking branch 'origin/master' into fsWriting

775cc49

use shuffleMemoryManager.pageSizeBytes

00804fe

asfgit closed this in 49702bd Aug 7, 2015

scwf mentioned this pull request Aug 8, 2015

[SPARK-8968] [SQL] external sort by the partition clomns when dynamic partitioning to optimize the memory overhead #7336

Closed

yjshen reviewed Aug 12, 2015
View reviewed changes

marmbrus deleted the fsWriting branch March 8, 2016 00:04

[SPARK-8890][SQL] Fallback on sorting when writing many dynamic partitions #8010

[SPARK-8890][SQL] Fallback on sorting when writing many dynamic partitions #8010

Uh oh!

Conversation

marmbrus commented Aug 6, 2015

Uh oh!

SparkQA commented Aug 6, 2015

Uh oh!

davies Aug 6, 2015

Choose a reason for hiding this comment

Uh oh!

SparkQA commented Aug 6, 2015

Uh oh!

SparkQA commented Aug 7, 2015

Uh oh!

rxin Aug 7, 2015

Choose a reason for hiding this comment

Uh oh!

liancheng commented Aug 7, 2015

Uh oh!

SparkQA commented Aug 7, 2015

Uh oh!

SparkQA commented Aug 7, 2015

Uh oh!

SparkQA commented Aug 7, 2015

Uh oh!

marmbrus commented Aug 7, 2015

Uh oh!

yjshen Aug 12, 2015

Choose a reason for hiding this comment

Uh oh!

marmbrus Aug 12, 2015

Choose a reason for hiding this comment

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

7 participants