Я объединяю два столбца в spark типа Array[Map[String,String]]
, в результате чего получается новый столбец типа 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, еще лучше
Адаптировано из ответа здесь. Seq
нужно вместо Array
.
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]]|
+--------------------+--------------------+----------------------------------------+
Обычный
Array
— это массив Java, на самом деле он не поддерживает все те же функции, что и Seq. Чтобы сделать это, существует неявное преобразование вWrappedArray
, но это не происходит автоматически внутри Spark, вы должны принудительно объявить этоSeq