将Array [string]类型的两个spark sql列合并到一个新的Array [string]列中

Question

我在Spark SQL中有两列DataFrame,每列中的每个条目都是一个字符串数组.

val  ngramDataFrame = Seq(
  (Seq("curious", "bought", "20"), Seq("iwa", "was", "asj"))
).toDF("filtered_words", "ngrams_array")

我想合并每行中的数组,以在新列中生成单个数组.我的代码如下:

def concat_array(firstarray: Array[String], 
                 secondarray: Array[String]) : Array[String] = 
                                     { (firstarray ++ secondarray).toArray }
val concatUDF = udf(concat_array _)
val concatFrame = ngramDataFrame.withColumn("full_array", concatUDF($"filtered_words", $"ngrams_array"))

我可以concat_array在两个数组上成功使用该函数.但是,当我运行上面的代码时,我得到以下异常:

org.apache.spark.SparkException:作业因阶段失败而中止:阶段16.0中的任务0失败1次,最近失败:阶段16.0中失去的任务0.0(TID 12,localhost):org.apache.spark.SparkException:失败在org.apache.spark.sql.execution的org.apache.spark.sql.catalyst.expressions.GeneratedClass $ GeneratedIterator.processNext(未知来源)执行用户定义的函数(anonfun $ 1 :(数组,数组)=>数组) .BufferedRowIterator.hasNext(BufferedRowIterator.java:43)at org.apache.spark.sql.execution.WholeStageCodegenExec $$ anonfun $ 8 $$ anon $ 1.hasNext(WholeStageCodegenExec.scala:370)at scala.collection.Iterator $$ anon $ 10 .hasNext(Iterator.scala:389)at sca.collection.Iterator $$ anon $ 11.hasNext(Iterator.scala:408)at org.apache.spark.shuffle.sort.BypassMergeSortShuffleWriter.write(BypassMergeSortShuffleWriter.java:125)at at Org.apache.spark.scheduler.ShuffleMapTask.runTask(ShuffleMapTask.scala:79)atg.apache.spark.scheduler.ShuffleMapTask.runTask(ShuffleMapTask.scala:47)at org.ap ache.spark.scheduler.Task.run(Task.scala:86)at org.apache.spark.executor.Executor $ TaskRunner.run(Executor.scala:274)at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor. java:1149)java.util.concurrent.ThreadPoolExecutor $ Worker.run(ThreadPoolExecutor.java:624)at java.lang.Thread.run(Thread.java:748)引起:java.lang.ClassCastException:scala.collection .mutable.WrappedArray $ ofRef无法强制转换为[Ljava.lang.String; 在$ line80.$ read $$ iw $$ iw $$ iw $$ iw $$ iw $$ iw $$ iw $$ iw $$ iw $$ iw $$ iw $$ iw $$ iw $$ iw $$ iw $$ iw $$ anonfun $ 1.apply(:76)... 13更多驱动程序堆栈跟踪:

Answer 1

在Spark 2.4或更高版本中,您可以使用concat(如果您想保留重复项):

ngramDataFrame.withColumn(
  "full_array", concat($"filtered_words", $"ngrams_array")
).show

+--------------------+---------------+--------------------+
|      filtered_words|   ngrams_array|          full_array|
+--------------------+---------------+--------------------+
|[curious, bought,...|[iwa, was, asj]|[curious, bought,...|
+--------------------+---------------+--------------------+

或array_union(如果你想删除重复):

ngramDataFrame.withColumn(
  "full_array",
   array_union($"filtered_words", $"ngrams_array")
)

例如,这些也可以由其他更高阶的函数组成

ngramDataFrame.withColumn(
   "full_array",
   flatten(array($"filtered_words", $"ngrams_array"))
)

与重复,和

ngramDataFrame.withColumn(
   "full_array",
   array_distinct(flatten(array($"filtered_words", $"ngrams_array")))
)

没有.

另外,WrappedArray在使用ArrayType列时不应使用.相反,你应该期待保证接口,这是Seq.所以udf应该使用具有以下签名的函数:

(Seq[String], Seq[String]) => Seq[String]

有关详细信息,请参阅SQL编程指南.

Answer 2

Arjun 您创建的 udf 中存在错误。当您传递数组类型列时。数据类型不是 Array[String]，而是 WrappedArray[String]。下面我将修改后的 udf 与输出一起粘贴。

val SparkCtxt = new SparkContext(sparkConf)

val sqlContext = new SQLContext(SparkCtxt)

import sqlContext.implicits

import org.apache.spark.sql.functions._
val temp=SparkCtxt.parallelize(Seq(Row(Array("String1","String2"),Array("String3","String4"))))
val df= sqlContext.createDataFrame(temp,
  StructType(List(
    StructField("Col1",ArrayType(StringType),true),
    StructField("Col2",ArrayType(StringType),true)
  )
  )    )

def concat_array(firstarray: mutable.WrappedArray[String],
                 secondarray: mutable.WrappedArray[String]) : mutable.WrappedArray[String] =
{
 (firstarray ++ secondarray)
}
val concatUDF = udf(concat_array _)
val df2=df.withColumn("udftest",concatUDF(df.col("Col1"), df.col("Col2")))
df2.select("udftest").foreach(each=>{println("***********")
println(each(0))})
df2.show(true)

输出：

+------------------+------------------+--------------------+
|              Col1|              Col2|             udftest|
+------------------+------------------+--------------------+
|[String1, String2]|[String3, String4]|[String1, String2...|
+------------------+------------------+--------------------+

WrappedArray(字符串1、字符串2、字符串3、字符串4)