无法在spark中将有序数据写入拼花

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

但是，当我查询文件时，我看到内容乱序。具体来说，“乱序”似乎更像是几个有序的数据帧分区已合并到文件中。

拼花行组元数据显示排序后的字段实际上是重叠的(例如，一个特定的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，查询可以在一个行组中找到它。

我已经尝试了上述代码的许多变体。例如，有和没有合并（我知道合并是不好的，但我的想法是用它来防止洗牌）。我也尝试了排序而不是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|

这看起来像是交错排序的分区。因此，我认为重新分区在这里没有给我带来任何好处，而且sort似乎无法保持写入步骤的顺序。

我读到我想做的事情应该是可能的。我甚至尝试过Ryan Blue的演讲“Parquet性能调优：缺失的指南”中概述的方法（不幸的是，它位于OReily付费墙后面）。这涉及使用insertInto。在这种情况下，spark似乎使用了旧版本的parquet mr，这损坏了元数据，我不确定如何升级它。

我不确定我做错了什么。我的感觉是我误解了重新分区（$“日期”）和排序工作和/或交互的方式。

我会感谢任何想法。为这篇文章道歉。:)

编辑:还要注意，如果我在< code>transformed.sort("id "，" sampleTime")上执行< code>show(n)操作，数据会正确排序。因此，问题似乎发生在写入阶段。如上所述，在写入过程中，排序的输出似乎被打乱了。

Answer 1

只是想法，合并后排序:"。合并(1)。sortWithinPartitions()"。预期结果看起来也很奇怪——为什么需要拼花地板中的有序数据？阅读后排序看起来更合适。

Answer 2

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

val requiredOrdering = partitionColumns ++ bucketIdExpression ++ sortColumns

其中分区列是用于对数据进行分区的列。您没有使用存储桶，因此存储桶 IdExpression 和 sortColumns 在此示例中不相关，所需的排序将仅是分区列。因此，如果这是您的代码：

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 将再次检查数据是否按日期排序，这次是（满足要求），因此 Spark 将保留此顺序，并且在保存数据时不再诱导排序。所以我相信这样它应该有效。