Spark-java.lang.ClassCastException，当在udf中处理Array[Array[Map[S

Question

我正在连接类型为Array[Map[String,String]]的spark中的两列，从而产生类型为Array[Array[Map[String,String]]]的新列。然而，我希望将该列展平，最终得到类型为Array[Map[String,String]]的列，其中包含两个原始列的值

我从Spark 2.4中了解到，可以将flatten直接应用于列的串联。类似这样的东西：

df.withColumn("concatenation", flatten(array($"colArrayMap1", $"colArrayMap2")))

然而，我仍然使用Spark 2.2，所以我需要使用udf。这是我写的：

def flatten_collection(arr: Array[Array[Map[String,String]]]) = {
if(arr == null)
null
else
arr.flatten
}

val flatten_collection_udf = udf(flatten_collection _)
df.withColumn("concatenation", array($"colArrayMap1", $"colArrayMap2")).withColumn("concatenation", flatten_collection_udf($"concatenation")).show(false)

但我得到了以下错误：

Caused by: org.apache.spark.SparkException: Failed to execute user defined function($anonfun$1: (array<array<map<string,string>>>) => array<map<string,string>>)
at org.apache.spark.sql.catalyst.expressions.GeneratedClass$GeneratedIterator.processNext(Unknown Source)
at org.apache.spark.sql.execution.BufferedRowIterator.hasNext(BufferedRowIterator.java:43)
at org.apache.spark.sql.execution.WholeStageCodegenExec$$anonfun$8$$anon$1.hasNext(WholeStageCodegenExec.scala:395)
at org.apache.spark.sql.execution.SparkPlan$$anonfun$2.apply(SparkPlan.scala:234)
at org.apache.spark.sql.execution.SparkPlan$$anonfun$2.apply(SparkPlan.scala:228)
at org.apache.spark.rdd.RDD$$anonfun$mapPartitionsInternal$1$$anonfun$apply$25.apply(RDD.scala:835)
at org.apache.spark.rdd.RDD$$anonfun$mapPartitionsInternal$1$$anonfun$apply$25.apply(RDD.scala:835)
at org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:49)
at org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:323)
at org.apache.spark.rdd.RDD.iterator(RDD.scala:287)
at org.apache.spark.scheduler.ResultTask.runTask(ResultTask.scala:87)
at org.apache.spark.scheduler.Task.run(Task.scala:109)
at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:380)
at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142)
at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)
at java.lang.Thread.run(Thread.java:748)
Caused by: java.lang.ClassCastException: scala.collection.mutable.WrappedArray$ofRef cannot be cast to [[Lscala.collection.immutable.Map;

我认为转换错误发生在udf中，但为什么以及如何避免它？

此外，如果有人知道Spark 2.2的解决方案，它不需要使用UDF，甚至更好的

Answer 1

改编自此处的答案。需要CCD_ 5而不是CCD_。

def concat_arr(
arr1: Seq[Map[String,String]],
arr2: Seq[Map[String,String]]
) : Seq[Map[String,String]] =
{
(arr1 ++ arr2)
}
val concatUDF = udf(concat_arr _)
val df2 = df.withColumn("concatenation", concatUDF($"colArrayMap1", $"colArrayMap2"))
df2.show(false)
+--------------------+--------------------+----------------------------------------+
|colArrayMap1        |colArrayMap2        |concatenation                           |
+--------------------+--------------------+----------------------------------------+
|[[a -> b], [c -> d]]|[[a -> b], [c -> d]]|[[a -> b], [c -> d], [a -> b], [c -> d]]|
+--------------------+--------------------+----------------------------------------+