收集并订购 DataFrame 列

Question

我有一个如下的DataFrame

A B C
1 3 1
1 8 2
1 5 3
2 2 1

我的输出应该是，B 列是根据 B 列的初始值排序的

A B
1 3,1/5,3/8,2
2 2,1

我写了这样的东西是scala

df.groupBy("A").withColumn("B",collect_list(concat("B",lit(","),"C"))

但是 dint 解决了我的问题。

Answer 1

鉴于您的输入数据框为

+---+---+---+
|A  |B  |C  |
+---+---+---+
|1  |3  |1  |
|1  |8  |2  |
|1  |5  |3  |
|2  |2  |1  |
+---+---+---+

你可以得到以下输出

+---+---------------+
|A  |B              |
+---+---------------+
|1  |[3,1, 5,3, 8,2]|
|2  |[2,1]          |
+---+---------------+

通过简单的groupBy、aggregations 和使用functions

df.orderBy("B").groupBy("A").agg(collect_list(concat_ws(",", col("B"), col("C"))) as "B")

您可以使用udf 函数来获得最终想要的结果

def joinString = udf((b: mutable.WrappedArray[String]) => {
  b.mkString("/")
} )

newdf.withColumn("B", joinString(col("B"))).show(false)

你应该得到

+---+-----------+
|A  |B          |
+---+-----------+
|1  |3,1/5,3/8,2|
|2  |2,1        |
+---+-----------+

注意您需要import org.apache.spark.sql.functions._ 才能使上述所有功能起作用

已编辑

B 列是根据 B 列的初始值排序的

为此，您可以将 orderBy 部分删除为

 import org.apache.spark.sql.functions._
val newdf = df.groupBy("A").agg(collect_list(concat_ws(",", col("B"), col("C"))) as "B")

def joinString = udf((b: mutable.WrappedArray[String]) => {
  b.mkString("/")
} )

newdf.withColumn("B", joinString(col("B"))).show(false)

你应该得到输出为

+---+-----------+
|A  |B          |
+---+-----------+
|1  |3,1/8,2/5,3|
|2  |2,1        |
+---+-----------+

Answer 2

这是你可以通过使用 concat_ws 函数然后 groupby 列 A 并收集列表来实现的

val df1 = spark.sparkContext.parallelize(Seq(
  ( 1, 3, 1),
  (1, 8, 2),
  (1, 5, 3),
  (2, 2, 1)
)).toDF("A", "B", "C")

val result = df1.withColumn("B", concat_ws("/", $"B", $"C"))

result.groupBy("A").agg(collect_list($"B").alias("B")).show

输出：

+---+---------------+
|  A|              B|
+---+---------------+
|  1|[3/1, 8/2, 5/3]|
|  2|          [2/1]|
+---+---------------+

已编辑： 如果要使用 B 列进行排序，可以执行以下操作

val format = udf((value : Seq[String]) => {
  value.sortBy(x => {x.split(",")(0)}).mkString("/")
})


val result = df1.withColumn("B", concat_ws(",", $"B", $"C"))
  .groupBy($"A").agg(collect_list($"B").alias("B"))
  .withColumn("B", format($"B"))

result.show()

输出：

+---+-----------+
|  A|          B|
+---+-----------+
|  1|3,1/5,3/8,2|
|  2|        2,1|
+---+-----------+

希望这对您有所帮助！