apache-spark - Spark SQL 替换 MySQL 的 GROUP_CONCAT 聚合函数

Question

我有一个包含两个字符串类型列的表 (username, friend)对于每个用户名，我想将其所有 friend 收集在一行中，并连接为字符串。例如:('username1', 'friends1, friends2, friends3')
我知道 MySQL 是用 GROUP_CONCAT 做到这一点的.有没有办法用 Spark SQL 做到这一点？

Answer 1

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

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

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: How to map Python with Scala or Java User Defined Functions? 所示。

在实践中提取 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"))