SPARK SQL замена для агрегатной функции mysql GROUP_CONCAT
У меня есть таблица из двух столбцов строкового типа (имя пользователя, друг), и для каждого имени пользователя я хочу собрать всех его друзей в одну строку, объединенные в строки ('username1', 'friends1, friends2, friends3'). Я знаю, что MySql делает это с помощью GROUP_CONCAT, есть ли способ сделать это с помощью SPARK SQL?
Спасибо
10 ответов
Прежде чем продолжить: эта операция является еще одной другой 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?
На практике это может быть быстрее, чтобы извлечь 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"))
В 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)
)
Выход:
+--------+--------------------------+
|username|friends |
+--------+--------------------------+
|jacques |nicolas, georges, francois|
|bob |zoe, amelie |
+--------+--------------------------+
Это похоже на MySQL GROUP_CONCAT() и Redshift LISTAGG().
Вы можете попробовать функцию 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")
Вот функция, которую вы можете использовать в 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
- искровое разрешение 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
Один из способов сделать это с 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"))
Язык: 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|
+---------+---------------+
Вы также можете использовать функцию Spark SQL collect_list, и после этого вам нужно будет привести к строке и использовать функцию regexp_replace для замены специальных символов.
regexp_replace(regexp_replace(regexp_replace(cast(collect_list((column)) as string), ' ', ''), ',', '|'), '[^A-Z0-9|]', '')
это более легкий способ.
Ниже код на основе Python, который обеспечивает функциональность group_concat.
Входные данные:
Cust_No,Cust_Cars
1, Toyota
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
Функция высшего порядка
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|
+-------+------------------+----------------+---------------------+