Искра создать Dataframe в UDF - PullRequest
Новая
       34

Искра создать Dataframe в UDF

0 голосов
/ 30 ноября 2018

У меня есть пример, я хочу создать Dataframe в UDF.Нечто похожее на приведенное ниже 

import org.apache.spark.ml.classification.LogisticRegressionModel
import org.apache.spark.ml.linalg.Vector
import org.apache.spark.ml.feature.VectorAssembler

данные в Dataframe 

    val df = Seq((1,1,34,23,34,56),(2,1,56,34,56,23),(3,0,34,23,23,78),(4,0,23,34,78,23),(5,1,56,23,23,12),
(6,1,67,34,56,34),(7,0,23,23,23,56),(8,0,12,34,45,89),(9,1,12,34,12,34),(10,0,12,34,23,34)).toDF("id","label","tag1","tag2","tag3","tag4")
    val assemblerDF = new VectorAssembler().setInputCols(Array("tag1", "tag2", "tag3","tag4")).setOutputCol("features")
    val data = assemblerDF.transform(df)
    val Array(train,test) = data.randomSplit(Array(0.6, 0.4), seed = 11L)
    val testData=test.toDF    

    val loadmodel=LogisticRegressionModel.load("/user/xu/savemodel")
    sc.broadcast(loadmodel)
    val assemblerFe = new VectorAssembler().setInputCols(Array("a", "b", "c","d")).setOutputCol("features")
    sc.broadcast(assemblerFe)

UDF 

    def predict(predictSet:Vector):Double={
        val set=Seq((1,2,3,4)).toDF("a","b","c","d")
        val predata = assemblerFe.transform(set)
        val result=loadmodel.transform(predata)
        result.rdd.take(1)(0)(3).toString.toDouble}

    spark.udf.register("predict", predict _)
    testData.registerTempTable("datatable")
    spark.sql("SELECT predict(features) FROM datatable").take(1)

я получаю ошибку вроде 

ERROR Executor: Exception in task 3.0 in stage 4.0 (TID 7) [Executor task launch worker for task 7]
org.apache.spark.SparkException: Failed to execute user defined function($anonfun$1: (vector) => double)

и 

WARN TaskSetManager: Lost task 3.0 in stage 4.0 (TID 7, localhost, executor driver): org.apache.spark.SparkException: Failed to execute user defined function($anonfun$1: (vector) => double)

Не поддерживается ли датафрейм?Я использую Spark 2.3.0 и Scala 2.11.спасибо

1 Ответ

0 голосов
/ 30 ноября 2018

Как уже упоминалось в комментариях, вам не нужен UDF для применения модели Trained для проверки данных.Вы можете применить модель для тестирования фрейма данных в самой основной программе, как показано ниже: 

val df = Seq((1,1,34,23,34,56),(2,1,56,34,56,23),(3,0,34,23,23,78),(4,0,23,34,78,23),(5,1,56,23,23,12),
(6,1,67,34,56,34),(7,0,23,23,23,56),(8,0,12,34,45,89),(9,1,12,34,12,34),(10,0,12,34,23,34)).toDF("id","label","tag1","tag2","tag3","tag4")
val assemblerDF = new VectorAssembler().setInputCols(Array("tag1", "tag2", "tag3","tag4")).setOutputCol("features")
val data = assemblerDF.transform(df)
val Array(train,test) = data.randomSplit(Array(0.6, 0.4), seed = 11L)
val testData=test.toDF    

val loadmodel=LogisticRegressionModel.load("/user/xu/savemodel")
sc.broadcast(loadmodel)
val assemblerFe = new VectorAssembler().setInputCols(Array("a", "b", "c","d")).setOutputCol("features")
sc.broadcast(assemblerFe)


val set=Seq((1,2,3,4)).toDF("a","b","c","d")
val predata = assemblerFe.transform(set)
val result=loadmodel.transform(predata) // Applying model on predata dataframe. You can apply model on any DataFrame.

Теперь результат - это Фрейм данных, вы можете повторно зарегистрировать Фрейм данных в виде таблицы и запросить предвидениеLabel и функции с помощью SQL ИЛИ вы можете напрямуювыберите предикат и другие поля в DataFrame.

Обратите внимание, что UDF - это функция Spark SQL для определения новых функций на основе столбцов, расширяющих словарь DSL Spark SQL для преобразования наборов данных.Он не возвращает сам DataFrame в качестве возвращаемого типа.и, как правило, не рекомендуется использовать UDF без необходимости, обратитесь к: https://jaceklaskowski.gitbooks.io/mastering-spark-sql/spark-sql-udfs-blackbox.html

...