收集行作为列表与按Apache Spark分组 [英] Collect rows as list with group by apache spark

问题描述

我有一个特殊的用例，其中我为同一位客户有多行，每行对象看起来像:

I have a particular use case where I have multiple rows for same customer where each row object looks like:

root
 -c1: BigInt
 -c2: String
 -c3: Double
 -c4: Double
 -c5: Map[String, Int]

现在，我按列c1进行分组，并为同一客户收集所有行作为列表，例如:

Now I have do group by column c1 and collect all the rows as list for same customer like:

c1, [Row1, Row3, Row4]
c2, [Row2, Row5]

我尝试过这种方式 dataset.withColumn("combined", array("c1","c2","c3","c4","c5")).groupBy("c1").agg(collect_list("combined"))，但出现异常:

I tried doing this ways dataset.withColumn("combined", array("c1","c2","c3","c4","c5")).groupBy("c1").agg(collect_list("combined")) but I get an exception:

Exception in thread "main" org.apache.spark.sql.AnalysisException: cannot resolve 'array(`c1`, `c2`, `c3`, `c4`, `c5`)' due to data type mismatch: input to function array should all be the same type, but it's [bigint, string, double, double, map<string,map<string,double>>];;

推荐答案

您可以使用struct函数代替array来合并列，并使用groupBy和collect_list聚合函数作为

Instead of array you can use struct function to combine the columns and use groupBy and collect_list aggregation function as

import org.apache.spark.sql.functions._
df.withColumn("combined", struct("c1","c2","c3","c4","c5"))
    .groupBy("c1").agg(collect_list("combined").as("combined_list"))
    .show(false)

，以便您将分组数据集与schema一起作为

root
 |-- c1: integer (nullable = false)
 |-- combined_list: array (nullable = true)
 |    |-- element: struct (containsNull = true)
 |    |    |-- c1: integer (nullable = false)
 |    |    |-- c2: string (nullable = true)
 |    |    |-- c3: string (nullable = true)
 |    |    |-- c4: string (nullable = true)
 |    |    |-- c5: map (nullable = true)
 |    |    |    |-- key: string
 |    |    |    |-- value: integer (valueContainsNull = false)

我希望答案会有所帮助

这篇关于收集行作为列表与按Apache Spark分组的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！