Spark SQL 取代 MySQL 的 GROUP_CONCAT 聚合函数

Question

我有一个包含两个字符串类型列(username, friend)的表，对于每个用户名，我想在一行中收集它的所有朋友，并将其连接为字符串。例如:('username1', 'friends1, friends2, friends3')

我知道MySQL这样做与GROUP_CONCAT。有什么办法做到这一点与Spark SQL?

Answer 1

在您继续之前:此操作是另一个groupByKey。虽然它有多个合法的应用程序，但它相对昂贵，所以请确保只在需要时使用它。

---

不完全是简洁或有效的解决方案，但您可以使用Spark 1.5.0中引入的UserDefinedAggregateFunction:

object GroupConcat extends UserDefinedAggregateFunction {
    def inputSchema = new StructType().add("x", StringType)
    def bufferSchema = new StructType().add("buff", ArrayType(StringType))
    def dataType = StringType
    def deterministic = true 
    def initialize(buffer: MutableAggregationBuffer) = {
      buffer.update(0, ArrayBuffer.empty[String])
    }
    def update(buffer: MutableAggregationBuffer, input: Row) = {
      if (!input.isNullAt(0)) 
        buffer.update(0, buffer.getSeq[String](0) :+ input.getString(0))
    }
    def merge(buffer1: MutableAggregationBuffer, buffer2: Row) = {
      buffer1.update(0, buffer1.getSeq[String](0) ++ buffer2.getSeq[String](0))
    }
    def evaluate(buffer: Row) = UTF8String.fromString(
      buffer.getSeq[String](0).mkString(","))
}

使用例子:

val df = sc.parallelize(Seq(
  ("username1", "friend1"),
  ("username1", "friend2"),
  ("username2", "friend1"),
  ("username2", "friend3")
)).toDF("username", "friend")
df.groupBy($"username").agg(GroupConcat($"friend")).show
## +---------+---------------+
## | username|        friends|
## +---------+---------------+
## |username1|friend1,friend2|
## |username2|friend1,friend3|
## +---------+---------------+

你也可以创建一个Python包装器，如Spark:如何映射Python与Scala或Java用户定义函数?

在实践中，提取RDD, groupByKey, mkString和重建DataFrame可以更快。

您可以通过将collect_list函数(Spark>= 1.6.0)与concat_ws:

组合来获得类似的效果

import org.apache.spark.sql.functions.{collect_list, udf, lit}
df.groupBy($"username")
  .agg(concat_ws(",", collect_list($"friend")).alias("friends"))

Answer 2

你可以试试collect_list函数

sqlContext.sql("select A, collect_list(B), collect_list(C) from Table1 group by A

或者你可以注册一个像

这样的UDF

sqlContext.udf.register("myzip",(a:Long,b:Long)=>(a+","+b))

你可以在查询

中使用这个函数

sqlConttext.sql("select A,collect_list(myzip(B,C)) from tbl group by A")

Answer 3

在Spark 2.4+中，有了collect_list()和array_join()的帮助，这变得更简单了。

下面是PySpark的演示，不过Scala的代码也应该非常相似:

from pyspark.sql.functions import array_join, collect_list
friends = spark.createDataFrame(
    [
        ('jacques', 'nicolas'),
        ('jacques', 'georges'),
        ('jacques', 'francois'),
        ('bob', 'amelie'),
        ('bob', 'zoe'),
    ],
    schema=['username', 'friend'],
)
(
    friends
    .orderBy('friend', ascending=False)
    .groupBy('username')
    .agg(
        array_join(
            collect_list('friend'),
            delimiter=', ',
        ).alias('friends')
    )
    .show(truncate=False)
)

在Spark SQL中，解决方案同样是:

SELECT
    username,
    array_join(collect_list(friend), ', ') AS friends
FROM friends
GROUP BY username;

输出:

+--------+--------------------------+
|username|friends                   |
+--------+--------------------------+
|jacques |nicolas, georges, francois|
|bob     |zoe, amelie               |
+--------+--------------------------+

这类似于MySQL的GROUP_CONCAT()和Redshift的LISTAGG()。

Answer 4

这是一个可以在PySpark中使用的函数:

import pyspark.sql.functions as F
def group_concat(col, distinct=False, sep=','):
    if distinct:
        collect = F.collect_set(col.cast(StringType()))
    else:
        collect = F.collect_list(col.cast(StringType()))
    return F.concat_ws(sep, collect)

table.groupby('username').agg(F.group_concat('friends').alias('friends'))

在SQL:

select username, concat_ws(',', collect_list(friends)) as friends
from table
group by username

Answer 5

—spark SQL解析与collect_set

SELECT id, concat_ws(', ', sort_array( collect_set(colors))) as csv_colors
FROM ( 
  VALUES ('A', 'green'),('A','yellow'),('B', 'blue'),('B','green') 
) as T (id, colors)
GROUP BY id

Answer 6

使用pyspark <1.6，不幸的是不支持用户定义的聚合函数:

byUsername = df.rdd.reduceByKey(lambda x, y: x + ", " + y)

如果你想让它再次成为一个数据框架:

sqlContext.createDataFrame(byUsername, ["username", "friends"])

从1.6开始，您可以使用collect_list，然后加入已创建的列表:

from pyspark.sql import functions as F
from pyspark.sql.types import StringType
join_ = F.udf(lambda x: ", ".join(x), StringType())
df.groupBy("username").agg(join_(F.collect_list("friend").alias("friends"))

Answer 7

语言: ScalaSpark版本: 1.5.2

我有同样的问题，也试图使用udfs来解决它，但不幸的是，这导致了更多的问题，后来在代码中由于类型不一致。我能够通过首先将DF转换为RDD，然后通过 分组并以所需的方式操纵数据，然后将RDD转换回DF来解决这个问题，如下所示:

val df = sc
     .parallelize(Seq(
        ("username1", "friend1"),
        ("username1", "friend2"),
        ("username2", "friend1"),
        ("username2", "friend3")))
     .toDF("username", "friend")
+---------+-------+
| username| friend|
+---------+-------+
|username1|friend1|
|username1|friend2|
|username2|friend1|
|username2|friend3|
+---------+-------+
val dfGRPD = df.map(Row => (Row(0), Row(1)))
     .groupByKey()
     .map{ case(username:String, groupOfFriends:Iterable[String]) => (username, groupOfFriends.mkString(","))}
     .toDF("username", "groupOfFriends")
+---------+---------------+
| username| groupOfFriends|
+---------+---------------+
|username1|friend2,friend1|
|username2|friend3,friend1|
+---------+---------------+

Answer 8

下面是实现group_concat功能的基于python的代码。

输入数据:

Cust_No, Cust_Cars

丰田

1,

2、宝马

1,奥迪

2、现代

from pyspark.sql import SparkSession
from pyspark.sql.types import StringType
from pyspark.sql.functions import udf
import pyspark.sql.functions as F
spark = SparkSession.builder.master('yarn').getOrCreate()
# Udf to join all list elements with "|"
def combine_cars(car_list,sep='|'):
  collect = sep.join(car_list)
  return collect
test_udf = udf(combine_cars,StringType())
car_list_per_customer.groupBy("Cust_No").agg(F.collect_list("Cust_Cars").alias("car_list")).select("Cust_No",test_udf("car_list").alias("Final_List")).show(20,False)

输出数据:Cust_No, Final_List

1,丰田|奥迪

2,宝马|现代

Answer 9

您也可以使用Spark SQL函数collect_list，之后您将需要转换为字符串并使用regexp_replace函数来替换特殊字符。

regexp_replace(regexp_replace(regexp_replace(cast(collect_list((column)) as string), ' ', ''), ',', '|'), '[^A-Z0-9|]', '')

Answer 10

高阶函数 concat_ws() 和 collect_list() 可以与 groupBy()

一起使用

import pyspark.sql.functions as F
    
df_grp = df.groupby("agg_col").agg(F.concat_ws("#;", F.collect_list(df.time)).alias("time"), F.concat_ws("#;", F.collect_list(df.status)).alias("status"), F.concat_ws("#;", F.collect_list(df.llamaType)).alias("llamaType"))

+-------+------------------+----------------+---------------------+
|agg_col|time              |status          |llamaType            |
+-------+------------------+----------------+---------------------+
|1      |5-1-2020#;6-2-2020|Running#;Sitting|red llama#;blue llama|
+-------+------------------+----------------+---------------------+