合并两个类型为Array[字符串]的列到一个新的Array[字符串]列

Question

Spark SQLDataFrame中有两列，每列中的每个条目都是字符串数组。

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函数。但是当我运行上述代码时，我得到以下异常：

组织。阿帕奇。火花SparkException:作业因阶段失败而中止：阶段16.0中的任务0失败1次，最近的失败：阶段16.0中的任务0.0丢失（TID 12，localhost）：org。阿帕奇。火花SparkException:无法执行用户定义的函数（anonfun$1:（数组，数组）=

Answer 1

Arjun您创建的udf中有一个错误。传递数组类型列时。数据类型不是数组[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（String1、String2、String3、String4）

Answer 2

在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")))
)

没有

另外，在处理ArrayType列时，不应使用WrappedArray。相反，您应该期望得到保证的接口，即Seq。因此，udf应该使用具有以下签名的函数：

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

详情请参阅SQL编程指南。