Замена SPARK SQL для функции агрегации mysql GROUP_CONCAT

Ответ 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: как сопоставить Python с функциями Scala или Java User Defined?

На практике может быть быстрее извлечь 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"))

Ответ 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")

Ответ 3

Вот функция, которую вы можете использовать в 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

Ответ 4

Один из способов сделать это с помощью pyspark & ​​lt; 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"))

Ответ 5

Язык: Scala Искра версия: 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|
+---------+---------------+

Ответ 6

Ниже приведен код на основе Python, обеспечивающий функциональность group_concat.

Входные данные:

Cust_No, Cust_Cars

1, Тойота

2, BMW

1, Audi

2, Hyundai

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, Toyota | Audi

2, BMW | Hyundai