Как вернуть ListBuffer в виде столбца из UDF, используя Spark Scala? - PullRequest
Новая
       31

Как вернуть ListBuffer в виде столбца из UDF, используя Spark Scala?

0 голосов
/ 26 сентября 2018

Я пытаюсь использовать UDF и возвращать ListBuffer в качестве столбца из UDF, получаю ошибку.

Я создал Df, выполнив код ниже: 

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

Я пытаюсь позвонить, выполнив следующий код: 

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 Ответ

0 голосов
/ 26 сентября 2018

Альтернатива с моими собственными вымышленными данными, которые вы можете адаптировать и не указывать UDF: 

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)
...