使用WithColumn與外部功能

Question

我有一個數據幀的數據與下面列

1）FILEFORMAT是以下所有列數據類型CSV 2）字符串

employeeid,pexpense,cexpense 

現在我需要建立一個新的數據幀其中有一個名爲expense的新列，它是根據列pexpense,cexpense進行計算的。

棘手的部分是計算算法是不是我創建了一個UDF功能，但它是一個外部函數，需要從一個Java庫，需要原始類型作爲參數輸入 - 在這種情況下pexpense，cexpense - 計算新列所需的值。

函數簽名是從外部Java罐子

public class MyJava 

{ 

    public Double calculateExpense(Double pexpense, Double cexpense) { 
     // calculation 
    } 

} 

所以，我怎麼可以調用外部函數來創建一個新的計算列。我可以在Spark應用程序中將該外部函數註冊爲UDF嗎？

Answer 1

您可以創建類似於以下（斑竹使用Scala的REPL）的外部方法的UDF：

// From a Linux shell prompt: 

vi MyJava.java 
public class MyJava { 
    public Double calculateExpense(Double pexpense, Double cexpense) { 
     return pexpense + cexpense; 
    } 
} 
:wq 

javac MyJava.java 
jar -cvf MyJava.jar MyJava.class 

spark-shell --jars /path/to/jar/MyJava.jar 

// From within the Spark shell 

val df = Seq(
    ("1", "1.0", "2.0"), ("2", "3.0", "4.0") 
).toDF("employeeid", "pexpense", "cexpense") 

val myJava = new MyJava 

val myJavaUdf = udf(
    myJava.calculateExpense _ 
) 

val df2 = df.withColumn("totalexpense", myJavaUdf($"pexpense", $"cexpense")) 

df2.show 
+----------+--------+--------+------------+ 
|employeeid|pexpense|cexpense|totalexpense| 
+----------+--------+--------+------------+ 
|   1|  1.0|  2.0|   3.0| 
|   2|  3.0|  4.0|   7.0| 
+----------+--------+--------+------------+ 

Answer 2

波紋管，是總和的一個例子兩列：

val somme= udf((a: Int, b: int) => a+b) 

val df_new = df.select(col("employeeid"), \ 
         col("pexpense"), \ 
         col("pexpense"), \ 
         somme(col("pexpense"), col("pexpense")) as "expense") 

Answer 3

通過使其作爲參數傳遞給該udf功能org.apache.spark.sql.functions可以簡單地「包裹」在UDF給定方法：

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

val myUdf = udf(calculateExpense _) 
val newDF = df.withColumn("expense", myUdf($"pexpense", $"cexpense")) 

此處假定pexpense和cexpense列均爲Double s。