我正在使用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，查询可以在一行组中找到它。
我已经尝试了上述代码的许多变体。例如，有和没有
coalesce（我知道合并是不好的，但是我的想法是使用它来防止混洗）。我还尝试了sort而不是sortWithinPartitions（sort应该创建一个总排序的排序，但是会导致很多分区）。例如：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的工作和/或交互方式。我将不胜感激。为这篇文章道歉。 :)
编辑：
另请注意，如果我在
transformed.sort("id", "sampleTime")上执行show（n），则数据将正确排序。因此，似乎该问题发生在写入阶段。如上所述，在写入过程中，排序的输出似乎已被改组。 最佳答案
问题在于，在保存文件格式时，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将保留此顺序，并且在保存数据时不会再进行排序。所以我相信这样应该可以。关于sorting - 无法将有序数据写入Spark中的 Parquet ，我们在Stack Overflow上找到一个类似的问题：https://stackoverflow.com/questions/52159938/