对partitionBy创建的一个输出目录中的数据进行排序

Question

我有一个很大的地理空间数据集partitionBy qk5的level 5。在每个quadkey级别的目录中，大约有1-50 Gb的数据，所以它不适合放在一个文件中。我想在进行地理空间查询时受益于下推过滤器。因此，我希望一个qk5分区中的文件按更高的qk分辨率排序(比方说四键级别10)。问:有没有一种方法可以在partitionBy批处理中对数据进行排序？例如：

qk5=00001/
    part1.parquet
    part2.parquet
    part3.parquet
    part4.parquet
...

qk5=33333/
    part10000.parquet
    part20000.parquet
    part30000.parquet
    part40000.parquet

我想让part1.parquet，part2.parquet，part3.parquite，part4.parquite中的数据按列'qk10‘排序。

下面是当前的代码，但它只提供了一个特定分区内的排序(例如part1.parket)：

// Parquet save
preExportRdd.toDF
  .repartition(partitionsNumber, $"salt")
  .sortWithinPartitions($"qk10")
  .drop("salt")
  .write
  .partitionBy("qk")
  .format("parquet")
  .option("compression", "gzip")
  .mode(SaveMode.Append)
  .save(exportUrl)

Answer 1

问题是你没有按照qk字段对数据帧进行全局排序，这会导致相同的qk值分布在不同的spark分区中。在写入阶段，由于partitionBy("qk")，写入特定物理分区(文件夹)的输出可能来自不同的spark分区，这会导致您的输出数据无法排序。

请尝试执行以下操作：

preExportRdd.toDF
  .repartitionByRange(partitionsNumber, $"qk", $"qk10", $"salt")
  .sortWithinPartitions($"qk10")
  .drop("salt")
  .write
  .partitionBy("qk")
  .format("parquet")
  .option("compression", "gzip")
  .mode(SaveMode.Append)
  .save(exportUrl)

repartitionByRange将根据提供的列对数据帧进行排序，并将排序后的数据帧拆分到所需数量的分区。