基于两列值对 Spark 数据集进行排序的有效方法是什么？

Question

我有一个包含以下格式的三列的大型数据集：

col1   col2   col3
------------------
 a1     1      i1
 a1     1      i2
 a1     2      i3
 a3     2      i4
 a3     1      i5
 a2     3      i6
 a2     3      i7
 a2     1      i8

我写了以下内容：

   val datase2 = dataset.groupBy("col1","col2").agg(collect_list("col3").as("col3"))
        .sort("col1", "col2")
        .groupBy("col1").agg(collect_list("col2"), collect_list("col3"))
        .toDF("col1", "col2", "col3").as[(String, Array[String], Array[String])]

为了从结果数据集中获取 col2 的不同值，我编写了以下内容：

dataset2.select("col3").distinct().show()

上面的代码适用于小型数据集，但对于大型数据集，我得到了以下类型的结果（只是为了说明结果数据集不一致的情况）：

col1     col2           col3
-----------------------------------
a1     [1, 2]      [[i1, i2], [i3]]
a2     [3, 1]      [[i6, i7], [i8]]
a3     [2, 1]      [[i4], [i5]]

正如我所做的那样sort("col1", "col2")输出应该是

col1     col2           col3
-----------------------------------
a1     [1, 2]      [[i1, i2], [i3]]
a2     [1, 3]      [[i8], [i6, i7]]
a3     [1, 2]      [[i5], [i4]]

col2 将按排序顺序排列，col2 和 col3 的值将根据其数组索引保持一致。例如，上述数据集的最后一行是

 col2           col3
-------------------------
[1, 2]      [[i5], [i4]]

但不是

 col2           col3
-------------------------
[1, 2]      [[i4], [i5]]

我怎样才能实现我的目标？

Answer 1

使用 struct 合并记录并使用 sort_array:

dataset
  .groupBy($"col1")
  .agg(sort_array(collect_list(struct($"col2", $"col3"))).alias("data"))
  .select($"col1", $"data.col2", $"data.col3")

致谢名单 user6910411) for 。