scala - 无法将有序数据写入 Spark 中的 Parquet

Question

我正在使用 Apache Spark 生成 Parquet 文件。我可以毫无问题地按日期对它们进行分区，但在内部我似乎无法按正确的顺序排列数据。
订单似乎在处理过程中丢失了，这意味着 Parquet 元数据不正确(特别是我想确保 Parquet 行组反射(reflect)排序顺序，以便特定于我的用例的查询可以通过元数据进行有效过滤)。
考虑以下示例:

// note: hbase source is a registered temp table generated from hbase
val transformed = sqlContext.sql(s"SELECT  id, sampleTime, ... , toDate(sampleTime) as date FROM hbaseSource")

// Repartion the input set by the date column (in my source there should be 2 distinct dates)
val sorted = transformed.repartition($"date").sortWithinPartitions("id", "sampleTime")

sorted.coalesce(1).write.partitionBy("date").parquet(s"/outputFiles")

通过这种方法，我确实得到了正确的 Parquet 分区结构(按日期)。更好的是，对于每个日期分区，我看到一个大的 Parquet 文件。

 /outputFiles/date=2018-01-01/part-00000-4f14286c-6e2c-464a-bd96-612178868263.snappy.parquet

但是，当我查询文件时，我看到内容乱序。具体来说，“乱序”似乎更像是几个有序的数据帧分区已合并到文件中。
Parquet 行组元数据显示排序的字段实际上是重叠的(例如，特定 id 可以位于许多行组中):

id:             :[min: 54, max: 65012, num_nulls: 0]
sampleTime:     :[min: 1514764810000000, max: 1514851190000000, num_nulls: 0]
id:             :[min: 827, max: 65470, num_nulls: 0]
sampleTime:     :[min: 1514764810000000, max: 1514851190000000, num_nulls: 0]
id:             :[min: 1629, max: 61412, num_nulls: 0]

我希望数据在每个文件中正确排序，以便每个行组中的元数据最小值/最大值不重叠。
例如，这是我想看到的模式:

RG 0: id:             :[min: 54, max: 100, num_nulls: 0]
RG 1: id:             :[min: 100, max: 200, num_nulls: 0]

...其中RG =“行组”。如果我想要 id = 75 ，查询可以在一个行组中找到它。
我已经尝试了上述代码的许多变体。例如有和没有 coalesce (我知道合并不好，但我的想法是用它来防止改组)。我也试过 sort而不是 sortWithinPartitions (排序应该创建一个完全有序的排序，但会导致许多分区)。例如:

val sorted = transformed.repartition($"date").sort("id", "sampleTime") 
sorted.write.partitionBy("date").parquet(s"/outputFiles")

给了我 200 个文件，太多了，而且它们仍然没有正确排序。我可以通过调整随机大小来减少文件数，但我希望在写入期间按顺序处理排序(我的印象是写入不会对输入进行随机排序)。我看到的顺序如下(为简洁起见省略了其他字段):

+----------+----------------+
|id|      sampleTime|
+----------+----------------+
|     56868|1514840220000000|
|     57834|1514785180000000|
|     56868|1514840220000000|
|     57834|1514785180000000|
|     56868|1514840220000000|

看起来它是交错排序的分区。所以我想 repartition在这里什么都不给我买，而且 sort似乎无法在写入步骤中保持顺序。
我读过我想做的事情应该是可能的。我什至尝试了演示文稿“Parquet 性能调整:
缺少的指南”由 Ryan Blue(不幸的是它在 OReily 付费墙后面)。这涉及使用 insertInto 。在这种情况下，spark 似乎使用了旧版本的 parquet-mr，它破坏了元数据，我不确定如何升级它。
我不确定我做错了什么。我的感觉是我误解了方式 repartition($"date")和 sort工作和/或互动。
我将不胜感激任何想法。为论文道歉。 :)
编辑:
另请注意，如果我执行 show(n)在 transformed.sort("id", "sampleTime")数据排序正确。 所以看起来问题发生在写入阶段。 如上所述，排序的输出似乎在写入过程中被打乱了。

Answer 1

问题是Spark在保存文件格式的时候是需要一定的顺序的，如果顺序不满足，Spark会在保存过程中根据要求对数据进行排序，忘记排序。更具体地说，Spark 需要这个顺序(这直接取自 Spark 2.4.4 的 Spark 源代码):

val requiredOrdering = partitionColumns ++ bucketIdExpression ++ sortColumns

哪里 partitionColumns是用于对数据进行分区的列。您没有使用分桶，所以 bucketingIdExpression和 sortColumns在这个例子和 requiredOrdering 中不相关将只是 partitionColumns .所以如果这是你的代码:

val sorted = transformed.repartition($"date").sortWithinPartitions("id", 
"sampleTime")

sorted.write.partitionBy("date").parquet(s"/outputFiles")

Spark 会检查数据是否按 date 排序，这不是，所以 Spark 会忘记你的排序并按 date 排序。 .另一方面，如果您改为这样做:

val sorted = transformed.repartition($"date").sortWithinPartitions("date", "id", 
"sampleTime")

sorted.write.partitionBy("date").parquet(s"/outputFiles")

Spark 会再次检查数据是否按 date 排序这一次是(满足要求)所以 Spark 将保留这个顺序，并且在保存数据时不会引入更多的排序。所以我相信这种方式应该有效。