IncompatibleSchemaException：意外的類型VectorUDT在Avro的格式序列化

Question

我使用星火Mllib來爲我的數據預測，然後將它們存儲到HDFS中的Avro格式時：

val dataPredictions = myModel.transform(myData) 
val output = dataPredictions.select("is", "probability", "prediction") 
output.write.format("com.databricks.spark.avro").save(path) 

我收到以下異常：

com.databricks.spark.avro.SchemaConverters$IncompatibleSchemaException: 
    Unexpected type org.apache.spark.ml.linalg.VectorUDT@3bfc3ba7. 

我的理解是'預測'列格式不能被序列化爲Avro。

• 如何將VectorUDT轉換爲數組，以便我可以在Avro中序列化它？
• 有沒有更好的選擇（我不能離開Avro格式）？

Answer 1

對任何Vector轉換爲Array[Double]，你可以使用下面的UDF：

import org.apache.spark.sql.functions.udf 
import org.apache.spark.sql.functions.col 
import org.apache.spark.ml.linalg.Vector 

val vectorToArrayUdf = udf((vector: Vector) => vector.toArray) 

// The following will work 
val output = dataPredictions 
    .withColumn("probabilities", vectorToArrayUdf(col("probability"))) 
    .select("id", "probabilities", "prediction") 

output.write.format("com.databricks.spark.avro").save(path)