UDF中的Spark classnotfoundexception

当我调用一个函数时，它就起作用了。但是当我在UDF中调用该函数将无法工作。UDF中的Spark classnotfoundexception

这是完整的代码。

val sparkConf = new SparkConf().setAppName("HiveFromSpark").set("spark.driver.allowMultipleContexts","true") 
val sc = new SparkContext(sparkConf) 
val hive = new org.apache.spark.sql.hive.HiveContext(sc) 

///////////// UDFS 
def toDoubleArrayFun(vec:Any) : scala.Array[Double] = { 
    return vec.asInstanceOf[WrappedArray[Double]].toArray 
} 
def toDoubleArray=udf((vec:Any) => toDoubleArrayFun(vec)) 

//////////// PROCESS 
var df = hive.sql("select vec from mst_wordvector_tapi_128dim where word='soccer'") 
println("==== test get value then transform") 
println(df.head().get(0)) 
println(toDoubleArrayFun(df.head().get(0))) 

println("==== test transform by udf") 
df.withColumn("word_v", toDoubleArray(col("vec"))) 
.show(10);

然后这个输出。

sc: org.apache.spark.SparkContext = [email protected] 
hive: org.apache.spark.sql.hive.HiveContext = 
toDoubleArrayFun: (vec: Any)Array[Double] 
toDoubleArray: org.apache.spark.sql.UserDefinedFunction 
df: org.apache.spark.sql.DataFrame = [vec: array<double>] 
==== test get value then transform 
WrappedArray(-0.88675,, 0.0216657) 
[[email protected] 
==== test transform by udf 
org.apache.spark.SparkException: Job aborted due to stage failure: Task 0 in stage 2.0 failed 4 times, most recent failure: Lost task 0.3 in stage 2.0 (TID 5, xdad008.band.nhnsystem.com): java.lang.ClassNotFoundException: $iwC$$iwC$$iwC$$iwC$$iwC$$$$5ba2a895f25683dd48fe725fd825a71$$$$$$iwC$$anonfun$toDoubleArray$1 
    at java.net.URLClassLoader.findClass(URLClassLoader.java:381) 
    at java.lang.ClassLoader.loadClass(ClassLoader.java:424) 
    at java.lang.ClassLoader.loadClass(ClassLoader.java:357)

此处输出完整。 https://gist.github.com/jeesim2/efb52f12d6cd4c1b255fd0c917411370

正如你所看到的“toDoubleArrayFun”函数运行良好，但在udf它声明ClassNotFoundException。

我无法更改配置单元数据结构，并且需要将vec转换为Array [Double]才能生成Vector实例。

那么上面的代码有什么问题？

星火版本是1.6.1

更新1

蜂巢表的 'VEC' 列类型为 “array<double>”

下面的代码也会导致错误

var df = hive.sql("select vec from mst_wordvector_tapi_128dim where 
word='hh'") 
df.printSchema() 
var word_vec = df.head().get(0) 
println(word_vec) 
println(Vectors.dense(word_vec))

输出

df: org.apache.spark.sql.DataFrame = [vec: array<double>] 
root 
|-- vec: array (nullable = true) 
| |-- element: double (containsNull = true) 
==== test get value then transform 
word_vec: Any = WrappedArray(-0.88675,...7) 
<console>:288: error: overloaded method value dense with alternatives: 
(values: Array[Double])org.apache.spark.mllib.linalg.Vector <and> 
(firstValue: Double,otherValues:Double*)org.apache.spark.mllib.linalg.Vector 
cannot be applied to (Any) 
println(Vectors.dense(word_vec))

这意味着配置单元'array<double>'列不能被铸造到Array<Double> 其实我想计算距离：双与两个array<double>列。如何添加基于array<double>列的矢量列？

典型方法是

Vectors.sqrt(Vectors.dense(Array<Double>, Array<Double>)

来源

2017-08-25 Jihun No

由于udf函数去序列化和反序列化过程，any数据类型将无法正常工作。您将必须定义传递给udf函数的列的确切数据类型。

从你的问题的输出似乎你有你的数据帧只有一列，即vec这是Array[Double]类型的

df: org.apache.spark.sql.DataFrame = [vec: array<double>]

其实没有必要的是UDF功能为您的vec列已经是的Array数据类型，这就是你的udf函数所做的，也就是将值转换为Array[Double]。

现在，你的其他函数调用工作

println(toDoubleArrayFun(df.head().get(0)))

，因为没有必要序列化和反序列化过程中，它只是斯卡拉函数调用。

来源

2017-08-25 10:07:49

啊，可序列化的重点！但是，我如何使数据帧列'Array '，而不是任何？我更新了问题！ –

而不是你的udf函数中的'Any'，只需将数据类型定义为'WrappedArray [Double]'，你应该没问题。 :) –

谢谢你的答案..顺便说一句，当我设置参数类型为'WrappedArray [双]'而不是'任何'它失败。 '：346：错误：类型不匹配;发现：任何。需要：scala.collection.mutable.WrappedArray [Double] println（Vectors.dense（toDoubleArrayFun（df.head（）。get（0））））' –

UDF中的Spark classnotfoundexception

回答

相关问题