試圖從UDF

Question

我想寫使用Scala的火花框架內聯函數，將一個字符串輸入，執行SQL語句並返回我一個字符串值執行火花SQL查詢

val testfunc: (String=>String)= (arg1:String) => 
{val k = sqlContext.sql("""select c_code from r_c_tbl where x_nm = "something" """)        
k.head().getString(0) 
} 

我註冊這個階函數作爲UDF

val testFunc_test = udf(testFunc) 

我有一個數據幀在蜂巢表

val df = sqlContext.table("some_table") 

然後我在withColumn中調用udf並嘗試將其保存在新的數據框中。

val new_df = df.withColumn("test", testFunc_test($"col1")) 

但每次我試圖做到這一點我得到一個錯誤

16/08/10 21:17:08 WARN TaskSetManager: Lost task 0.0 in stage 1.0 (TID 1,  10.0.1.5): java.lang.NullPointerException 
    at org.apache.spark.sql.execution.SQLExecution$.withNewExecutionId(SQLExecution.scala:41) 
    at org.apache.spark.sql.DataFrame.withNewExecutionId(DataFrame.scala:2086) 
    at org.apache.spark.sql.DataFrame.foreach(DataFrame.scala:1434) 

我是比較新的火花和Scala。但我不知道爲什麼這個代碼不應該運行。任何見解或工作將受到高度讚賞。

請注意，我沒有粘貼整個錯誤堆棧。請讓我知道是否需要。

Answer 1

您的UDF中不能使用sqlContext - UDF必須是可序列化的才能發送給執行程序，並且上下文（可以認爲是到羣集的連接）不能被序列化併發送到節點 - 只有驅動程序應用程序（其中UDF是定義爲，但不是執行）可以使用sqlContext。

看起來像你的用例（從表Y中的每個記錄的表X執行選擇）最好通過使用join來完成。