代码之家 › 专栏 › 技术社区 › Sai

如何使用Spark Scala从UDF返回ListBuffer作为列?

user-defined-functions apache-spark-sql apache-spark scala

Sai · 技术社区 · 7 年前

我试图使用UDF并将ListBuffer作为列从UDF返回,我得到了一个错误。

val df = Seq((1,"dept3@@rama@@kumar","dept3##rama#@kumar"), (2,"dept31@@rama1##kumar1","dept33##rama3#@kumar3")).toDF("id","str1","str2")
df.show()

它显示如下:

+---+--------------------+--------------------+
| id|                str1|                str2|
+---+--------------------+--------------------+
|  1|  dept3@@rama@@kumar|  dept3##rama#@kumar|
|  2|dept31@@rama1##ku...|dept33##rama3#@ku...|
+---+--------------------+--------------------+

根据我的要求,我必须使用我必须根据一些输入拆分上述列,因此我尝试了如下UDF:

    def appendDelimiterError=udf((id: Int, str1: String, str2: String)=> {
            var lit = new ListBuffer[Any]()
            if(str1.contains("@@"){val a=str1.split("@@")}
            else if(str1.contains("##"){val a=str1.split("##")}
            else if(str1.contains("#&"){val a=str1.split("#&")}
            if(str2.contains("@@"){ val b=str2.split("@@")}
            else if(str2.contains("##"){ val b=str2.split("##") }
            else if(str1.contains("#@"){val b=str2.split("#@")}
            var tmp_row = List(a,"test1",b)
            lit +=tmp_row 
return lit 
})

我尝试通过执行以下代码来校准:

val df1=df.appendDelimiterError("newcol",appendDelimiterError(df("id"),df("str1"),df("str2"))

我收到错误“这是一个错误的呼叫”。我想使用ListBuffer/list存储并返回呼叫地点。

我的预期产出将是:

+---+--------------------+------------------------+----------------------------------------------------------------------+ 
| id|                str1|                str2    |                               newcol                                 |
+---+--------------------+------------------------+----------------------------------------------------------------------+
|  1|  dept3@@rama@@kumar|  dept3##rama#@kumar    |ListBuffer(List("dept","rama","kumar"),List("dept3","rama","kumar"))  |
|  2|dept31@@rama1##kumar1|dept33##rama3#@kumar3  | ListBuffer(List("dept31","rama1","kumar1"),List("dept33","rama3","kumar3")) |                                          
+---+--------------------+------------------------+----------------------------------------------------------------------+

如何做到这一点?

1 回复 | 直到 7 年前

Ged 7 年前

另一种选择是我自己的虚构数据,您可以根据这些数据进行定制,而无需自定义项:

import org.apache.spark.sql.functions.{col, udf}  
import org.apache.spark.sql.expressions._
import org.apache.spark.sql.functions._

val df = Seq(
  (1, "111@#cat@@666", "222@@fritz@@777"),
  (2, "AAA@@cat@@555", "BBB@@felix@@888"),
  (3, "HHH@@mouse@@yyy", "123##mickey#@ZZZ") 
 ).toDF("c0", "c1", "c2")

 val df2 = df.withColumn( "c_split", split(col("c1"), ("(@#)|(@@)|(##)|(#@)")  ))
          .union(df.withColumn("c_split", split(col("c2"), ("(@#)|(@@)|(##)|(#@)")  )) )
 df2.show(false)
 df2.printSchema()


 val df3 = df2.groupBy(col("c0")).agg(collect_list(col("c_split")).as("List_of_Data") )   
 df3.show(false)
 df3.printSchema()

给出答案,但没有ListBuffer-真的有必要吗?如下所示:

+---+---------------+----------------+------------------+
|c0 |c1             |c2              |c_split           |
+---+---------------+----------------+------------------+
|1  |111@#cat@@666  |222@@fritz@@777 |[111, cat, 666]   |
|2  |AAA@@cat@@555  |BBB@@felix@@888 |[AAA, cat, 555]   |
|3  |HHH@@mouse@@yyy|123##mickey#@ZZZ|[HHH, mouse, yyy] |
|1  |111@#cat@@666  |222@@fritz@@777 |[222, fritz, 777] |
|2  |AAA@@cat@@555  |BBB@@felix@@888 |[BBB, felix, 888] |
|3  |HHH@@mouse@@yyy|123##mickey#@ZZZ|[123, mickey, ZZZ]|
+---+---------------+----------------+------------------+

root
 |-- c0: integer (nullable = false)
 |-- c1: string (nullable = true)
 |-- c2: string (nullable = true)
 |-- c_split: array (nullable = true)
 |    |-- element: string (containsNull = true)

 +---+---------------------------------------+
 |c0 |List_of_Data                           |
 +---+---------------------------------------+
 |1  |[[111, cat, 666], [222, fritz, 777]]   |
 |3  |[[HHH, mouse, yyy], [123, mickey, ZZZ]]|
 |2  |[[AAA, cat, 555], [BBB, felix, 888]]   |
 +---+---------------------------------------+

 root
  |-- c0: integer (nullable = false)
  |-- List_of_Data: array (nullable = true)
  |    |-- element: array (containsNull = true)
  |    |    |-- element: string (containsNull = true)