无法将有序数据写入火花中的镶木地板

Question

我正在使用 Apache Spark 生成镶木地板文件。我可以毫无问题地按日期对它们进行分区，但在内部我似乎无法按正确的顺序排列数据。

订单似乎在处理过程中丢失了，这意味着镶木地板元数据不正确（特别是我想确保镶木地板行组反映排序顺序，以便特定于我的用例的查询可以通过元数据进行有效过滤）。

考虑以下示例：

// 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")

通过这种方法，我确实得到了正确的镶木地板分区结构（按日期）。更好的是，对于每个日期分区，我看到一个大的镶木地板文件。

 /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（排序应该创建一个完全有序的排序，但会导致很多分区）。例如：

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

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

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

看起来它是交错排序的分区。所以我认为repartition在这里没有给我买任何东西，而且sort似乎无法在写入步骤中保持顺序。

我读过我想做的事情应该是可能的。我什至尝试了 Ryan Blue 的演示文稿“Parquet 性能调整：缺少的指南”中概述的方法（不幸的是，它位于 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 将保留此顺序，并且在保存数据时不会再进行排序。所以我相信这种方式应该有效。

Answer 2

只是想法，合并后排序：“.coalesce(1).sortWithinPartitions()”。另外预期的结果看起来很奇怪 - 为什么需要镶木地板中的有序数据？看完之后排序看起来更合适。