[SPARK-12539][SQL] support writing bucketed table

[cloud-fan]

This PR adds bucket write support to Spark SQL. User can specify bucketing columns, numBuckets and sorting columns with or without partition columns. For example:

df.write.partitionBy("year").bucketBy(8, "country").sortBy("amount").saveAsTable("sales")

When bucketing is used, we will calculate bucket id for each record, and group the records by bucket id. For each group, we will create a file with bucket id in its name, and write data into it. For each bucket file, if sorting columns are specified, the data will be sorted before write.

Note that there may be multiply files for one bucket, as the data is distributed.

Currently we store the bucket metadata at hive metastore in a non-hive-compatible way. We use different bucketing hash function compared to hive, so we can't be compatible anyway.

Limitations:

• Can't write bucketed data without hive metastore.
• Can't insert bucketed data into existing hive tables.

[cloud-fan]

cc @yhuai@nongli

[SparkQA]

Test build #48367 has finished for PR 10498 at commit 8cb2494.

• This patch passes all tests.
• This patch merges cleanly.
• This patch adds the following public classes (experimental):\n * case class Hash(children: Seq[Expression]) extends Expression\n * class TextOutputWriter(\n

[yhuai]

This one also includes #10435, right?

[nongli]

@cloud-fan

currently we don't shuffle before writing partitioned data, which means we will have same partition data in different RDD blocks, and that's why we have multi-files for one partition, and we will also have multi-files for one bucket, is that safe?

This is safe but how can we get in this state from a single write. There must have been a partitionBy before right?

hive support having bucketing without partitioning, should we support it?

Why not? If this hard to support?

[rxin]

BTW in github you can use square brackets to create a checklist, e.g.

- [ ] item a
- [ ] item b

becomes

• item a
• item b

[cloud-fan]

This one also includes #10435, we can merge that first.

[SparkQA]

Test build #48415 has finished for PR 10498 at commit a9dc997.

• This patch passes all tests.
• This patch merges cleanly.
• This patch adds the following public classes (experimental):
  • case class BucketSpec(

[nongli]

and the @SInCE annotations. and comments. These are public APIs

[liancheng]

This method can be simplified to:

if (sortingColumns.isDefined) {
require(numBuckets.isDefined, "sortBy must be used together with bucketBy")
}
for {
n <- numBuckets
cols <- normalizedBucketCols
} yield {
require(n >0, "Bucket number must be greater than 0.")
BucketSpec(n, cols, normalizedSortCols)
}

(require throws IllegalArgumentException when the condition is not met.)

[SparkQA]

Test build #48489 has finished for PR 10498 at commit d2dc9b3.

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

[cloud-fan]

I'm a little worried here. It was a simple != operator for non-bucket path before, but now it's a function call.

[nongli]

I think this is fine. The rest of this path is much more expensive than this function call.

[SparkQA]

Test build #48854 has finished for PR 10498 at commit d3200cf.

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

[cloud-fan]

Shall we put the bucket id at the very last? i.e. after the file extension, so that it's much easier to get the bucket id given a file name. e.g. part-r-00009-ea518ad4-455a-4431-b471-d24e03814677.gz.parquet.00002

cc @nongli

[nongli]

I don't have a strong opinion here. Let's go either way for now and talk another pass before shipping this. We should try this with hive as well just to get another data point.

[davies]

Having the bucket id at the very last could break some other applications, that rely on the file extension (to recognized the file format), so don't do that.

[SparkQA]

Test build #48856 has finished for PR 10498 at commit 1afd3ee.

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

[rxin]

I'd remove this TODO.

[nongli]

This looks good to me to merge.

[davies]

@cloud-fan Can we write bucketed table without partitions?

Just saw you have a test case for that, but didn't see you update the DefaultWriterContainer, how can that work?

[davies]

minor question: Why do we support bucketing for json writer? The bucketing can only be recognized by Spark SQL, in this case, parquet is much more efficient.

As we embed CSV into Spark, I think we don't need to support bucketing for that as well.

[cloud-fan]

just because we can... I felt it's cheap to add bucketing support for JSON so I went for it.

[rxin]

I'm going to merge this. @cloud-fan can you create a follow-up pr to address some of the comments above?

[l15k4]

Guys do you have a rough guess about when bucketing is to be implemented for org.apache.spark.sql.DataFrameWriter#save ?

[infinitymittal]

Hi,

There is the limitation of "Can't insert bucketed data into existing hive tables.". Do we have any plans to relax the same? I want to insert data using a query into an already existing table.

@cloud-fan Do we have a Jira for the same?

[tejasapatil]

@infinitymittal : See https://issues.apache.org/jira/browse/SPARK-17729

[infinitymittal]

@tejasapatil Thanks for the response. SPARK-17729 says "Spark still won't produce bucketed data as per Hive's bucketing guarantees". I want the data to be bucketed when written. Any further leads?

Just to be clear, even with "hive.enforce.bucketing" set to true, the data won't be written. Is that correct? Referencing pull request 15300's comments "Added test to ensure that INSERTs fail if strict bucket / sort is enforced".

[tejasapatil]

@infinitymittal : It will take time to have a fully functional support added. I had initiated a design proposal to get consensus on this could be done : https://issues.apache.org/jira/browse/SPARK-19256

In Spark, "hive.enforce.bucketing" is not respected. #15300 won't guarantee that the data written adheres to Hive's bucketing spec so approach taken there is to fail in user sets configs to enforce bucketing. This will avoid wrong data being written when user is expecting correct outputs after setting "hive.enforce.bucketing" to true. The longer term plan is to get rid of these configs and always write properly bucketed data (hive 2.x follows this model).

[FelixKJose]

@tejasapatil Is there any update on this regarding "always write properly bucketed data (hive 2.x follows this model)". Does spark provides this or your MR is ready to be merged into master?