[SPARK-3007][SQL]Add "Dynamic Partition" support to Spark Sql hive #1919

baishuo · 2014-08-13T09:14:59Z

the detail please refer the comment of https://issues.apache.org/jira/browse/SPARK-3007

AmplabJenkins · 2014-08-13T09:18:15Z

Can one of the admins verify this patch?

baishuo · 2014-08-13T09:18:23Z

I didnt add the related test since I dont know how to write it. can any one give me some instruction?:)
but I had test the function by SparkSQLCLIDriver and "sbt/sbt catalyst/test sql/test hive/test" passed

marmbrus · 2014-08-13T22:58:42Z

There are a couple of ways we can add tests, ideally we would do a little of both:

Find existing hive tests that test dynamic partitioning and add them to our whitelist. The test harness will automatically invoke hive to calculate the correct answers. You need to make sure you have Hadoop and Hive compiled and the environment variables set correctly as described in other dependencies for developers.
Add tests to HiveQuerySuite.

liancheng · 2014-08-18T06:39:17Z

sql/hive/src/main/scala/org/apache/spark/SparkHadoopWriter.scala

@@ -93,6 +93,33 @@ private[hive] class SparkHiveHadoopWriter(
      null)
  }

+  def open(dynamicPartPath: String) {
+    val numfmt = NumberFormat.getInstance()


NumberFormat.getInstance() is not thread-safe. We can use a thread-local variable to hold this object, similar to Cast.threadLocalDateFormat

Just realized this function is a variant of the original open() method within the same file. This should be a bug in the master branch.

Another issue is that, SparkHadoopWriter resides in project core, which is an indirect dependency of sql/hive. Thus logically, it's not proper to put open(dynamicPartPath: String) here.

Oh, it is actually SparkHiveHadoopWriter in sql/hive. Seems we need to rename this file.

liancheng · 2014-08-18T07:18:47Z

@yhuai It would be nice if you can have a look at this PR since you're the expert here :)

@baishuo You can refer to sql/README.md for details about setting up testing environment.

liancheng · 2014-08-18T07:24:11Z

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/Cast.scala

@@ -271,4 +272,9 @@ object Cast {
      new SimpleDateFormat("yyyy-MM-dd HH:mm:ss")
    }
  }
+  private[sql] val threadLocalNumberFormat = new ThreadLocal[NumberFormat] {


Ah, sorry, I didn't make myself clear enough. I mean you can refer to Cast.threadLocalDateFormat, not add the thread-local version of NumberFormat here, since it's not related to Cast. A better place to hold this could be object SparkHadoopWriter.

liancheng · 2014-08-18T07:43:39Z

Please don't forget to add golden answer files for those test cases newly added to whitelist in HiveCompatibilitySuite.

liancheng · 2014-08-18T09:14:16Z

sql/hive/src/main/scala/org/apache/spark/sql/hive/execution/InsertIntoHiveTable.scala

+          count += 1
+          writer2.write(record)
+        }
+        for((k,v) <- writerMap) {


Space before (

yhuai · 2014-08-18T15:34:15Z

@baishuo Thank you for working on it.

I have three general comments.

Hive has a lots of confs that are used to influence how semantic analyzer works. HiveConf.ConfVars.DYNAMICPARTITIONING (hive.exec.dynamic.partition) and HiveConf.ConfVars.DYNAMICPARTITIONINGMODE (hive.exec.dynamic.partition.mode) are two examples. As long as we generate the correct results and we can make sure the execution is robust, I think it is not necessarily to follow those confs.
For hive.exec.dynamic.partition.mode, I think the purpose of it is to avoid having too many concurrent file writers in a task. Actually, even if hive.exec.dynamic.partition.mode=strict, we can still have many distinct values on those dynamic partitioning columns and thus, have too many file writers in a task. For those columnar file formats, like RCFile, ORC, and Parquet, every file writer internally maintain a memory buffer. Many file writers can significantly increase the memory footprint of a task and can introduce OOMs. Instead of relying on Hive's confs, it is better to provide a way to group data based on those dynamic partitioning columns. So, we will not have many concurrent file writers. Just two primitive ideas. We can shuffle the data before inserting. Or, we can do local grouping and write data in a group-by-group fashion. Anyway, I feel we may need to introduce changes to the planner.
The last comment is not quite related to this PR. I think it is better to have a general design on how table is partitioned and (hopefully,) Hive's directory layout in HDFS will be just a special case. I am not sure that creating a single file for every combination of values of partitioning columns is a good way. It introduces potential stability issues to the insert operation (too many file writers), and performance issues to both insert and table scan operations. With this approach, we can easily create a lots of small files in HDFS, which introduces memory pressure to the HDFS namenode.

baishuo · 2014-08-19T02:39:49Z

thanks a lot @yhuai and @liancheng :)

baishuo · 2014-08-19T17:41:55Z

Hi @marmbrus and @liancheng I had made some modification and do the test with "sbt/sbt catalyst/test sql/test hive/test" . Please help me to check if it is proper when you have time . Thank you :)

liancheng · 2014-08-20T04:14:24Z

Hmm, I see 17 newly whitelisted test cases, but only golden answers for the dynamic_partition case were submitted.

baishuo · 2014-08-20T05:08:38Z

I also curious about that.
I down the master branch,and check the folder sql/hive/src/test/resources/golden
I find that files begin with dynamic_partition_skip_default* or load_dyn_part* already exist.

baishuo · 2014-08-20T10:31:04Z

Here I try to explain my design idea(the code is mostly in InsertIntoHiveTable.scala) :
lets assume there is a table called table1,which has 2 columns:col1,col2, and two partitions: part1, part2.

first:
In case of just insert data to a static partition,I find when "saveAsHiveFile" finished, the data was wroten to a temporary location, then directory like: /tmp/hive-root/hive_****/-ext-10000,lets call it TMPLOCATION. And under TMPLOCATION, there is sub directory /part1=.../part2=... , all data was store under TMPLOCATION/part1=.../part2=... , then spark will call hive api "loadPartition" to move the files to {hivewarehouse}/{tablename}/part1=.../part2=... and update the metadata. then the whole progress is OK.

If we what to implement the "dynamic partiton function", we need to use hive api "loadDynamicPartitions" to move data and update metadata. But the requirement of directory formate for "loadDynamicPartitions" is a little difference to "loadPartition":

1: In case of one static partition and one dynamic partition (HQL like "
insert overwrite table table1 partition(part1=val1,part2) select a,b,c from ..."), loadDynamicPartitions need the tmp data located at TMPLOCATION/part2=c1(there is NO "part1=val1", in the progress of loadDynamicPartitions, it wiil be added), TMPLOCATION/part2=c2 ......., And loadDynamicPartitions will move them to {hivewarehouse}/{tablename}/part1=val1/part2=c1, {hivewarehouse}/{tablename}/part1=val1/part2=c2 ...., and update the metadata. Note that in this case loadDynamicPartitions do note need the subdir like part1=val1 under TMPLOCATION

2: In case of zero static partition and 2 dynamic partition (HQL like "
insert overwrite table table1 partition(part1,part2) select a,b,x,c from ..."), loadDynamicPartitions need the tmp data located at TMPLOCATION/part1=../part2=c1, TMPLOCATION/part1=../part2=c2 ......., And loadDynamicPartitions will move them to {hivewarehouse}/{tablename}/part1=../part2=...,

So whether there is static partition in HQL determines how we create subdir under TMPLOCATION. That why the function "getDynamicPartDir" exist.

second:
where shall we call the "getDynamicPartDir"? must a location that we can get the values for dynamic partiton. so we call this function at "iter.map { row =>..." in the closure of "val rdd = childRdd.mapPartitions". when we get the row, we can get the values for dynamic partiton. after we get the dynamicPartPath by function getDynamicPartDir, we can pass it to next RDD by the output this RDD: serializer.serialize(outputData, standardOI) -> dynamicPartPath. (for the static partiton,dynamicPartPath is null)

when the next rdd (closure in writeToFile) get the data and dynamicPartPath, we can check if the dynamicPartPath equals null. if not null. we check if there is already a corresponding writer exist in writerMap which store all writer for each partition. if there is. we use this writer to write the record. that ensure the data belongs to same partition will be wroten to the same directory.

loadDynamicPartitions require there is no other files under TMPLOCATION except the subdir for dynamic partition. that why there are several "if (dynamicPartNum == 0)" in writeToFile

baishuo · 2014-08-26T08:31:44Z

Hi @marmbrus i had update the file relating with test. all test passed on my machine. Would you please help to verify this patch when you have time:) I had write out the thinking of the code. thank you.
@rxin @liancheng

marmbrus · 2014-08-26T18:55:25Z

Thanks for working on this! We will have more time to review it after the Spark 1.1 release.

marmbrus · 2014-08-30T00:15:22Z

ok to test

SparkQA · 2014-08-30T00:19:26Z

QA tests have started for PR 1919 at commit 0c324be.

This patch merges cleanly.

SparkQA · 2014-08-30T00:20:29Z

QA tests have finished for PR 1919 at commit 0c324be.

This patch fails unit tests.
This patch merges cleanly.
This patch adds the following public classes (experimental):
- case class Sqrt(child: Expression) extends UnaryExpression
- class TreeNodeRef(val obj: TreeNode[_])

baishuo · 2014-09-01T08:06:42Z

Hi @marmbrus , can you help me to check why the test fail? I had compile and do the test locally so I had thought it can passed the Spark QA test:) . And there is a new PR(with same changes, had test locally )#2226 base on new master. Would you please do a test on it if this PR still fails? thank you :)

liancheng · 2014-09-02T19:51:19Z

@baishuo Scala style check failed. See here for details.

liancheng · 2014-09-03T00:19:47Z

Would you mind to close this PR since #2226 was opened as a replacement?

baishuo · 2014-09-03T02:17:21Z

no problem， close this PR

@yhuai

a new PR base on new master. changes are the same as #1919 Author: baishuo(白硕) <vc_java@hotmail.com> Author: baishuo <vc_java@hotmail.com> Author: Cheng Lian <lian.cs.zju@gmail.com> Closes #2226 from baishuo/patch-3007 and squashes the following commits: e69ce88 [Cheng Lian] Adds tests to verify dynamic partitioning folder layout b20a3dc [Cheng Lian] Addresses @yhuai's comments 096bbbc [baishuo(白硕)] Merge pull request #1 from liancheng/refactor-dp 1093c20 [Cheng Lian] Adds more tests 5004542 [Cheng Lian] Minor refactoring fae9eff [Cheng Lian] Refactors InsertIntoHiveTable to a Command 528e84c [Cheng Lian] Fixes typo in test name, regenerated golden answer files c464b26 [Cheng Lian] Refactors dynamic partitioning support 5033928 [baishuo] pass check style 2201c75 [baishuo] use HiveConf.DEFAULTPARTITIONNAME to replace hive.exec.default.partition.name b47c9bf [baishuo] modify according micheal's advice c3ab36d [baishuo] modify for some bad indentation 7ce2d9f [baishuo] modify code to pass scala style checks 37c1c43 [baishuo] delete a empty else branch 66e33fc [baishuo] do a little modify 88d0110 [baishuo] update file after test a3961d9 [baishuo(白硕)] Update Cast.scala f7467d0 [baishuo(白硕)] Update InsertIntoHiveTable.scala c1a59dd [baishuo(白硕)] Update Cast.scala 0e18496 [baishuo(白硕)] Update HiveQuerySuite.scala 60f70aa [baishuo(白硕)] Update InsertIntoHiveTable.scala 0a50db9 [baishuo(白硕)] Update HiveCompatibilitySuite.scala 491c7d0 [baishuo(白硕)] Update InsertIntoHiveTable.scala a2374a8 [baishuo(白硕)] Update InsertIntoHiveTable.scala 701a814 [baishuo(白硕)] Update SparkHadoopWriter.scala dc24c41 [baishuo(白硕)] Update HiveQl.scala

Co-authored-by: Szehon Ho <szehon.apache@gmail.com>

baishuo added 4 commits August 13, 2014 01:27

Update HiveQl.scala

d3e206e

Update SparkHadoopWriter.scala

b22857a

Update InsertIntoHiveTable.scala

bade51d

Update InsertIntoHiveTable.scala

d211d33

baishuo added 3 commits August 18, 2014 14:29

Update HiveCompatibilitySuite.scala

f0f620d

Update InsertIntoHiveTable.scala

412a48b

Update HiveQuerySuite.scala

567972c

liancheng reviewed Aug 18, 2014
View reviewed changes

Update Cast.scala

8e51a4b

liancheng reviewed Aug 18, 2014
View reviewed changes

baishuo added 2 commits August 19, 2014 09:01

update file after test

af8411a

do a little modify

0c324be

baishuo mentioned this pull request Sep 1, 2014

[SPARK-3007][SQL]Add Dynamic Partition support to Spark Sql hive #2226

Closed

baishuo closed this Sep 3, 2014

snmvaughan pushed a commit to snmvaughan/spark that referenced this pull request Mar 26, 2024

rdar://124295468 : Bump Iceberg to 1.3.0.11-apple (apache#1919)

ab1489c

Co-authored-by: Szehon Ho <szehon.apache@gmail.com>

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

[SPARK-3007][SQL]Add "Dynamic Partition" support to Spark Sql hive #1919

[SPARK-3007][SQL]Add "Dynamic Partition" support to Spark Sql hive #1919

baishuo commented Aug 13, 2014

AmplabJenkins commented Aug 13, 2014

baishuo commented Aug 13, 2014

marmbrus commented Aug 13, 2014

liancheng Aug 18, 2014

liancheng Aug 18, 2014

yhuai Aug 18, 2014

liancheng commented Aug 18, 2014

liancheng Aug 18, 2014

liancheng commented Aug 18, 2014

liancheng Aug 18, 2014

yhuai commented Aug 18, 2014

baishuo commented Aug 19, 2014

baishuo commented Aug 19, 2014

liancheng commented Aug 20, 2014

baishuo commented Aug 20, 2014

baishuo commented Aug 20, 2014

baishuo commented Aug 26, 2014

marmbrus commented Aug 26, 2014

marmbrus commented Aug 30, 2014

SparkQA commented Aug 30, 2014

SparkQA commented Aug 30, 2014

baishuo commented Sep 1, 2014

liancheng commented Sep 2, 2014

liancheng commented Sep 3, 2014

baishuo commented Sep 3, 2014

[SPARK-3007][SQL]Add "Dynamic Partition" support to Spark Sql hive #1919

[SPARK-3007][SQL]Add "Dynamic Partition" support to Spark Sql hive #1919

Conversation

baishuo commented Aug 13, 2014

AmplabJenkins commented Aug 13, 2014

baishuo commented Aug 13, 2014

marmbrus commented Aug 13, 2014

liancheng Aug 18, 2014

Choose a reason for hiding this comment

liancheng Aug 18, 2014

Choose a reason for hiding this comment

yhuai Aug 18, 2014

Choose a reason for hiding this comment

liancheng commented Aug 18, 2014

liancheng Aug 18, 2014

Choose a reason for hiding this comment

liancheng commented Aug 18, 2014

liancheng Aug 18, 2014

Choose a reason for hiding this comment

yhuai commented Aug 18, 2014

baishuo commented Aug 19, 2014

baishuo commented Aug 19, 2014

liancheng commented Aug 20, 2014

baishuo commented Aug 20, 2014

baishuo commented Aug 20, 2014

baishuo commented Aug 26, 2014

marmbrus commented Aug 26, 2014

marmbrus commented Aug 30, 2014

SparkQA commented Aug 30, 2014

SparkQA commented Aug 30, 2014

baishuo commented Sep 1, 2014

liancheng commented Sep 2, 2014

liancheng commented Sep 3, 2014

baishuo commented Sep 3, 2014