Как сплющить упакованный массив в набор искровых данных в Java

Question

Использование Spark 2.2 Java 1.8

У меня есть требование к collect_set столбца массива. Но это дает мне WrappedArray. Пожалуйста, смотрите ниже.

Dataset<Row> df2 = df.groupBy("id").agg(collect_list("values"))
df2.show(truncate=False)
# +-----+----------------------------------------------+ 
# |id|                         collect_list(values) | 
# +-----+----------------------------------------------+ 
# |1    |[WrappedArray(1, 2, 3), WrappedArray(4, 5, 6)]| 
# |2    |[WrappedArray(2), WrappedArray(3)]            | 
# +-----+----------------------------------------------+

Expected output : = 

# +-----+------------------+
# |store|           values |
# +-----+------------------+
# |1    |[1, 2, 3, 4, 5, 6]|
# |2    |[2, 3]            |
# +-----+------------------+

Как мы можем достичь вышеупомянутого выхода в искровой Java. Может ли кто-нибудь помочь в этом? Спасибо.

Answer 1

Вот эквивалент скалы с использованием UDF (не java Guy):

//df.show(false)

+-----+----------------------------------------------+
|store|values                                        |
+-----+----------------------------------------------+
|1    |[WrappedArray(1, 2, 3), WrappedArray(4, 5, 6)]|
|2    |[WrappedArray(2), WrappedArray(3)]            |
+-----+----------------------------------------------+

import org.apache.spark.sql._
import org.apache.spark.sql.functions._

val flattenWrappedArray = udf((value:  Seq[Seq[Int]]) => {value.flatten})
df.withColumn("values_new",flattenWrappedArray($"values")).show(false)

Вывод:

+-----+----------------------------------------------+-------------+
|store|values                                        |values_new   |
+-----+----------------------------------------------+-------------+
|1    |[WrappedArray(1, 2, 3), WrappedArray(4, 5, 6)]|[1,2,3,4,5,6]|
|2    |[WrappedArray(2), WrappedArray(3)]            |[2,3]        |
+-----+----------------------------------------------+-------------+    

Надеюсь, это поможет!

Answer 2

Перед группированием можно использовать функцию "взорвать":

df.withColumn("values", explode($"values")).groupBy("id").agg(collect_list($"values"))