将两个Array [string]类型的spark sql列合并到新的Array [string]列中 [英] Merge two spark sql columns of type Array[string] into a new Array[string] column

问题描述

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

I have two columns in a Spark SQL DataFrame with each entry in either column as an array of strings.

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

我想合并每一行中的数组以在新列中创建一个数组.我的代码如下:

I want to merge the arrays in each row to make a single array in a new column. My code is as follows:

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函数.但是，当我运行上面的代码时，出现以下异常:

I can successfully use the concat_array function on two arrays. However when I run the above code, I get the following exception:

org.apache.spark.SparkException:由于阶段失败而导致作业中止:16.0阶段中的任务0失败了1次，最近一次失败:16.0阶段中的任务0.0(TID 12，本地主机)丢失了:org.apache.spark. SparkException:无法在org.apache.spark上的org.apache.spark.sql.catalyst.expressions.GeneratedClass $ GeneratedIterator.processNext(未知源)执行用户定义的函数(anonfun $ 1:((数组，数组)=>数组). sql.execution.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)在scala.collection.Iterator $$ anon $ 11.hasNext(Iterator.scala:408)在org.apache.spark.shuffle.sort.BypassMergeSortShuffleWriter.write(BypassMergeSortShuffleWriter.java: 125)位于org.org.apache.spark.scheduler.ShuffleMapTask.scala:79)org.org.apache.spark.scheduler.ShuffleMapTask.runTask(ShuffleMapTask.scala:47) org.apache.spark.executor.Executor $ TaskRunner.run(Executor.scala:274)处的java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor)处的.apache.spark.scheduler.Task.run(Task.scala:86) .java:1149)at 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 $ iw $ iw $$ iw $$ iw $$ anonfun $ 1.apply(:76)...还有13个驱动程序stacktrace:

org.apache.spark.SparkException: Job aborted due to stage failure: Task 0 in stage 16.0 failed 1 times, most recent failure: Lost task 0.0 in stage 16.0 (TID 12, localhost): org.apache.spark.SparkException: Failed to execute user defined function(anonfun$1: (array, array) => array) 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:370) at scala.collection.Iterator$$anon$10.hasNext(Iterator.scala:389) at scala.collection.Iterator$$anon$11.hasNext(Iterator.scala:408) at org.apache.spark.shuffle.sort.BypassMergeSortShuffleWriter.write(BypassMergeSortShuffleWriter.java:125) at org.apache.spark.scheduler.ShuffleMapTask.runTask(ShuffleMapTask.scala:79) at org.apache.spark.scheduler.ShuffleMapTask.runTask(ShuffleMapTask.scala:47) at org.apache.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) at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624) at java.lang.Thread.run(Thread.java:748) Caused by: java.lang.ClassCastException: scala.collection.mutable.WrappedArray$ofRef cannot be cast to [Ljava.lang.String; at $line80.$read$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$anonfun$1.apply(:76) ... 13 more Driver stacktrace:

推荐答案

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

Arjun there is an error in the udf you had created.when you are passing the array type columns .data type is not Array[String] it is WrappedArray[String].below i am pasting the modified udf along with output.

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(String1，String2，String3，String4)

WrappedArray(String1, String2, String3, String4)

这篇关于将两个Array [string]类型的spark sql列合并到新的Array [string]列中的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！