Scala在断开字符串后返回UDF中的多列

Question

我正在尝试分解一个字符串（从技术上讲，是从数据帧中的列传递的字符串），并将这些损坏的字符串作为列表返回到数据帧。斯卡拉版本 2.11。我更喜欢带有udf的斯卡拉或pyspark解决方案 - 因为udf内部发生了很多事情。

假设我有一个数据框架:

val df = List(("123", "a*b*c*d*e*f*x*y*z"), ("124", "g*h*i*j*k*l*m*n*o")).toDF("A", "B")

我想要的结果（在udf中，因为那里发生了很多事情;斯卡拉版本 2.11） --

 A       B
123    ((a, b, c),
        (d, e, f),
        (x, y, z))
124    ((g, h, i),
        (j, k, l), 
        (m, n, o))

编写一个udf来打破这个并返回列表-但是，我不知道如何定义或传递模式以将结果作为三列返回到数据帧中。

def testUdf =  udf( (s: String) => { 
  val a = s.split("\\*").take(3).toList
  val b = s.split("\\*").drop(3).take(3).toList
  val c = s.split("\\*").drop(6).take(3).toList
  val abc = (a, b, c).zipped.toList.asInstanceOf[List[String]]
  // println (abc) // This does not work
} )
val df2 = df.select($"A", testUdf($"B").as("B")) // does not work because of type mismatch. 

我尝试过这样做-但是，我不知道如何将模式传递给上面的Udf：

   val schema = StructType(List(
     StructField("C1", StringType),
     StructField("C2", StringType),
     StructField("C3", StringType)
   ))

此外，在此之后，我希望按照Spark SQL表中分解多个列的步骤来分解数据帧。

我们将非常感谢您的帮助。

Answer 1

问题是你的UDF返回< code>Unit(最后一个语句是返回值)。我建议采取以下步骤:

val df = List(("123", "a*b*c*d*e*f*x*y*z"), ("124", "g*h*i*j*k*l*m*n*o")).toDF("A", "B")

def testUdf = udf((s: String) => {
  val Array(s1, s2, s3, s4, s5, s6, s7, s8, s9) = s.split(s"\\*")
  Seq(
    (s1, s2, s3),
    (s4, s5, s6),
    (s7, s8, s9)
  )
})

val df2 = df.select($"A", explode(testUdf($"B")).as("B"))

df2.show()

+---+-------+
|  A|      B|
+---+-------+
|123|[a,b,c]|
|123|[d,e,f]|
|123|[x,y,z]|
|124|[g,h,i]|
|124|[j,k,l]|
|124|[m,n,o]|
+---+-------+

Answer 2

您在压缩之前生成数组的方式不会正确呈现元素。按所需顺序生成元素的一种方法是使用二维数组在应用压缩之前预转置元素。

以下 UDF 将 1） 将字符串列拆分为一个数组，该数组被转置为 2-D 数组，2） 将 2-D 数组的行压缩为元组数组，以及 3） 将元组数组转换为元组元组（即结构的列类型结构）：

val df = Seq(
  ("123", "a*b*c*d*e*f*x*y*z"),
  ("124", "g*h*i*j*k*l*m*n*o")
).toDF("A", "B")

import org.apache.spark.sql.functions._

def splitUdf = udf( (s: String) => {
  val arr = s.split("\\*")
  val arr2d = Array.ofDim[String](3, 3)

  for {
    r <- 0 until 3
    c <- 0 until 3
  } arr2d(r)(c) = arr(c * 3 + r)

  val arrTup = (arr2d(0), arr2d(1), arr2d(2)).zipped.toArray

  (arrTup(0), arrTup(1), arrTup(2))
} )

val df2 = df.select($"A", splitUdf($"B").as("B"))

df2.show(false)
// +---+-------------------------+
// |A  |B                        |
// +---+-------------------------+
// |123|[[a,b,c],[d,e,f],[x,y,z]]|
// |124|[[g,h,i],[j,k,l],[m,n,o]]|
// +---+-------------------------+

Answer 3

您定义的 udf 是从字符串到单位 - 从最后一行中删除 abc 以返回它 另外请注意，asInstanceOf[] 不会更改类型 - 您仍然有一个元组 下面将为您提供列表列表列表

def testUdf =  udf( (s: String) => { 
  val a = s.split("\\*").take(3).toList
  val b = s.split("\\*").drop(3).take(3).toList
  val c = s.split("\\*").drop(6).take(3).toList
  (a, b, c).zipped.toList.map(t=>List(t._1,t._2,t._3))
} )