apache-spark - 如何在 Spark SQL 中强制进行内存中的分块排序？

Question

Parquet 文件格式对记录的顺序很敏感。根据排序顺序，其柱状编码可能会产生明显更小的文件。 另一方面，对 TB 的输入记录进行排序非常昂贵。

假设 10GB 分成 block 允许内存中排序，同时生成几乎与整个 1TB 完全排序一样小的 Parquet 文件。

是否可以指示 Spark SQL 在生成 parquet 文件之前进行分块排序？

另一个用例是在编写统一的 Parquet 文件之前使用分块排序将许多小的 Parquet 文件合并为一个文件。

Answer 1

据我所知，Spark < 2.0.0 中没有这样的开箱即用选项。您可以尝试的一件事是在编写之前将 coalesce 与 Hive SORT BY 子句结合起来，这应该具有类似的效果:

val df: DataFrame = ???
val n: Int = ??? //

df.coalesce(n)
df.coalesce(n).registerTempTable("df")
sqlContext.sql("SELECT * FROM df SORT BY foo, bar").write.parquet(...)

或

df.coalesce(n).sortWithinPartitions($"foo", $"bar").write.parquet(...)

请记住，SORT BY 不等同于 DataFrame.sort。

Spark 2.0.0 引入了 sortBy 和 bucketBy 方法，其中后者按给定的列对每个桶中的输出进行排序 和 should support Parquet :

val df: DataFrame = ???
val nBuckets: Int = ???

df.write.bucketBy(nBuckets, "foo").sortBy("foo", "bar").saveAsTable(...)

注意:这似乎仅在使用 saveAsTable 保存 Parquet 文件时有效，但它看起来并不直接支持 parquet writer (df.write. spark-2.0.0-preview 中的 bucketBy(...).sortBy(...).parquet(...))。