如何为将来的排序合并连接保留排序的拼花表？答案

【问题标题】：How to persist sorted parquet tables for future sort merge joins?如何为将来的排序合并连接保留排序的拼花表？
【发布时间】：2015-11-19 22:34:55
【问题描述】：

我想在 S3 上将一个大型排序表持久化到 Parquet，然后将其读入并使用 Sorted Merge Join 策略将其连接到另一个大型排序表。

问题是：即使我事先根据连接键对这些表进行了排序，但一旦我将它们持久保存到 Parquet，它们似乎会丢失有关它们排序的信息。无论如何，是否有暗示 Spark 下次我阅读它们时不需要使用它们？

我一直在 Spark 1.5 上尝试这个，我不断得到 SQL EXPLAIN 计划看起来像：

[== Physical Plan ==]
[TungstenProject [pos#28400,workf...#28399]]
[ SortMergeJoin [CHROM#28403,pos#28400], [CHROM#28399,pos#28332]]
[  TungstenSort [CHROM#28403 ASC,pos#28400 ASC], false, 0]
[   TungstenExchange hashpartitioning(CHROM#28403,pos#28400)]
[    ConvertToUnsafe]
[     Scan ParquetRelation[file:/....sorted.parquet][pos#284....8424]]
[  TungstenSort [CHROM#28399 ASC,pos#28332 ASC], false, 0]
[   TungstenExchange hashpartitioning(CHROM#28399,pos#28332)]
[    ConvertToUnsafe]
[     Scan ParquetRelation[file:....exploded_sorted.parquet][pos#2.....399]]

您可以在其中看到额外的 TungstenExchange 和 TungstenSort 阶段，即使此联接位于两个表上，而这两个表在保存到 Parquet 之前已按联接键按 orderBy 排序。

【问题讨论】：

标签： apache-spark apache-spark-sql parquet

【解决方案1】：

看起来像这样is coming in Spark 2.0 along with support for bucketing。

【讨论】：

【解决方案2】：

不幸的是，Spark-2.0 还不支持通过分桶写入 S3。我昨天试过Spark-2.0-priview。

val NUMBER_OF_BUCKETS = 20
rdd.toDF.write.mode(SaveMode.Overwrite)
        .bucketBy(NUMBER_OF_BUCKETS,"data_frame_key")
        .partitionBy("day")
        .save("s3://XXXXX")

并收到此错误消息：

java.lang.IllegalArgumentException: Currently we don't support writing bucketed data to this data source.
    at org.apache.spark.sql.DataFrameWriter.assertNotBucketed(DataFrameWriter.scala:462)
    at org.apache.spark.sql.DataFrameWriter.save(DataFrameWriter.scala:243)

【讨论】：