处理过程中似乎会丢失顺序,这意味着parquet元数据不正确(具体来说,我希望确保parquet行组反映排序顺序,以便特定于我的用例的查询可以通过元数据有效地进行过滤)。
请考虑以下示例:
// 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
(我知道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
似乎无法在写入时保持顺序。
我已经读过可以实现我想要的功能。我甚至尝试过 Ryan Blue 在演示文稿《Parquet 性能调优:缺失的指南》中提出的方法(不幸的是它在 OReily 支付墙后面)。那种情况下,Spark 似乎使用了一个旧版本的 Parquet-mr,导致元数据损坏,而我不确定如何升级它。
我不确定我做错了什么。我的感觉是我误解了 repartition($"date")
和 sort
的工作方式和/或交互方式。
我会感激任何想法。对于这篇长文,我表示歉意。 :)
编辑:
还要注意,如果我在 transformed.sort("id", "sampleTime")
上执行 show(n)
,数据将被正确排序。所以问题似乎出现在写入阶段。正如上面所提到的,看起来排序的输出在写入时被洗牌了。