Erstellen Sie SparkSQL-UDF mit nicht serialisierbaren Objekten

Question

Ich versuche, eine UDF zu schreiben, die ich auf Hive-Tabellen in einem SQL-Context verwenden möchte. Ist es in irgendeiner Weise möglich, Objekte aus anderen Bibliotheken, die nicht serialisierbar sind, einzubeziehen? Hier ist ein minimales Beispiel dafür, was nicht funktioniert:

def myUDF(s: String) = { 
import sun.misc.BASE64Encoder 
val coder= new BASE64Encoder 
val encoded= decoder.encode(s) 
encoded 
} 

ich registrieren, um die Funktion in den Funken Shell als UDF-Funktion

val encoding = sqlContext.udf.register("encoder", myUDF) 

Wenn ich versuche es auf einen Tisch „test“ zu laufen

sqlContext.sql("SELECT encoder(colname) from test").show() 

ich erhalte den Fehler

org.apache.spark.SparkException: Task not serializable 
object not serializable (class: sun.misc.BASE64Encoder, value: sun.misc.BASE64Encoder@4a7f9a94) 

Gibt es einen Workaround dafür? Ich habe versucht, myUDF in ein Objekt und in eine Klasse einzubetten, aber das hat auch nicht funktioniert.

Answer 1

können Sie versuchen, die Definition udf Funktion als

def encoder = udf((s: String) => { 
    import sun.misc.BASE64Encoder 
    val coder= new BASE64Encoder 
    val encoded= coder.encode(s.getBytes("UTF-8")) 
    encoded 
}) 

und rufen Sie die udf Funktion als

dataframe.withColumn("encoded", encoder(col("id"))).show 

Aktualisiert

Wie @santon klargestellt hat, dass BASE64Encoder Encoder für eingeleitet jede Zeile im Datenrahmen, was zu Leistungsproblemen führen kann. Die Lösung dazu wäre, ein statisches ObjektBASE64Encoder zu erstellen und es innerhalb udf Funktion zu benennen.