UDF - Aufgabe nicht serializable Ausnahme

Question

Ich versuche UDF zu erstellen mit dem unter scala Code

lazy val formattedDF = df.withColumn("result_col", validateudf(df("id"))) 

val validateudf = udf((id: Int) => { 

    if(id == 1){ 
    "ID IS EQUAL TO 1" 
    } 
    else if(id > 1){ 
    validateId(id) 
    } 
    else{ 
    "NO VALID RECORDS" 
    } 
}) 

def validateId(id:Int) : String = { 
    if (id > 2) { 
    "ID IS GREATER THAN 2" 
    } 
    else { 
    "VALID RECORDS" 
    } 
} 

Wenn ich diesen Code ausführen, erhalte ich Aufgabe nicht serializable Ausnahme.

Irgendwelche Ideen? Vielen Dank.

Answer 1

udf wird als Blackbox genommen, die die Säulen geleitet erfordert werden serialisiert und deserialisiert so verwenden von udf wird nicht empfohlen, bis Sie Alternativen mit eingebauten Funktionen haben.

Aufruf udf Funktion mit withColumn ist in Ordnung, aber Sie haben eine andere Funktion validateId aus dem Innern der udf Funktion aufgerufen, die das Problem verursacht.

Ich würde vorschlagen, Sie nicht udf Funktion überhaupt zu verwenden, wie Sie die Anforderung nur mithilfe von when eingebaute Funktion erreichen können.

import org.apache.spark.sql.functions._ 
val formattedDF2 = df.withColumn("result_col", when($"id" === 1, lit("ID IS EQUAL TO 1")).otherwise(when($"id" > 2, lit("ID IS GREATER THAN 2")).otherwise(when($"id" > 1, lit("VALID RECORDS")).otherwise(lit("NO VALID RECORDS"))))) 

Ich hoffe, die Antwort hilfreich