scala - 将 DataFrame 写为 CSV 并按列分区时，如何保留 DataFrame 的顺序？

Question

我对 DataFrame 的行进行排序，然后像这样将其写入磁盘:

df.
  orderBy("foo").
  write.
  partitionBy("bar", "moo").
  option("compression", "gzip").
  csv(outDir)

当我查看生成的 .csv.gz 文件时，它们的顺序没有保留。这是 Spark 的做法吗？将 DF 写入带有分区的磁盘时，有没有办法保持顺序？

编辑:更准确地说:不是 CSV 的顺序不正确，而是它们内部的顺序。假设我在 df.orderBy 之后有如下内容(为简单起见，我现在只按一列进行分区):

foo | bar | baz
===============
  1 |   1 |   1
  1 |   2 |   2
  1 |   1 |   3
  2 |   3 |   4
  2 |   1 |   5
  3 |   2 |   6
  3 |   3 |   7
  4 |   2 |   9
  4 |   1 |  10

我希望它是这样的，例如对于文件夹 bar=1 中的文件:

part-00000-NNN.csv.gz:

1,1
1,3
2,5

part-00001-NNN.csv.gz:

3,8
4,10

但它是什么样的:

part-00000-NNN.csv.gz:

1,1
2,5
1,3

part-00001-NNN.csv.gz:

4,10
3,8

Answer 1

已经有一段时间了，但我又亲眼目睹了这一点。我终于找到了解决方法。

假设，您的架构如下:

• 时间:bigint
• channel :字符串
• 值(value):双倍

如果你这样做:

df.sortBy("time").write.partitionBy("channel").csv("hdfs:///foo")

单个 part-* 文件中的时间戳会被丢弃。

如果你这样做:

df.sortBy("channel", "time").write.partitionBy("channel").csv("hdfs:///foo")

顺序正确。

我认为这与洗牌有关。因此，作为一种变通方法，我现在首先按要对数据进行分区的列进行排序，然后按我希望在各个文件中对其进行排序的列进行排序。