Funken udf mit Datenrahmen

Question

Ich benutze Spark 1.3. Ich habe ein Dataset, in dem die Daten in der Spalte (Spalte Bestelldatum) im Format JJJJ/MM/TT vorliegen. Ich möchte einige Berechnungen mit Daten machen und deshalb möchte ich jodatime verwenden, um einige Konvertierungen/Formatierungen durchzuführen. Hier ist die udf, die ich habe:

val return_date = udf((str: String, dtf: DateTimeFormatter) => dtf.formatted(str)) 

Hier ist der Code, wo der udf aufgerufen wird. Ich bekomme jedoch einen Fehler mit der Angabe "Nicht zutreffend". Muss ich diese UDF registrieren oder fehlt mir hier etwas?

val user_with_dates_formatted = users.withColumn(
    "formatted_date", 
    return_date(users("ordering_date"), DateTimeFormat.forPattern("yyyy/MM/dd") 
) 

Answer 1

Ich glaube nicht, dass Sie in den DateTimeFormatter als Argument an die UDF passieren können. Sie können nur eine Column übergeben. Eine Lösung wäre, zu tun:

val return_date = udf((str: String, format: String) => { 
    DateTimeFormat.forPatten(format).formatted(str)) 
}) 

Und dann:

val user_with_dates_formatted = users.withColumn(
    "formatted_date", 
    return_date(users("ordering_date"), lit("yyyy/MM/dd")) 
) 

Ehrlich gesagt, obwohl - sowohl dies als auch Ihre ursprünglichen Algorithmen das gleiche Problem haben. Sie analysieren beide yyyy/MM/dd mit forPattern für jeden Datensatz. Besser wäre ein Singleton-Objekt um einen Map[String,DateTimeFormatter], vielleicht wie diese gewickelt zu erstellen (gründlich ungetestet, aber Sie erhalten die Idee):

object DateFormatters { 
    var formatters = Map[String,DateTimeFormatter]() 

    def getFormatter(format: String) : DateTimeFormatter = { 
    if (formatters.get(format).isEmpty) { 
     formatters = formatters + (format -> DateTimeFormat.forPattern(format)) 
    } 
    formatters.get(format).get 
    } 
} 

Dann würden Sie Ihre UDF ändern:

val return_date = udf((str: String, format: String) => { 
    DateFormatters.getFormatter(format).formatted(str)) 
}) 

That Weg, DateTimeFormat.forPattern(...) wird nur einmal pro Format pro Executor aufgerufen.

Eine Sache, über die Singleton-Objekt-Lösung zu beachten ist, dass Sie nicht das Objekt in der spark-shell definieren können - Sie haben es in einer JAR-Datei zu packen und nutzen die --jars Option spark-shell, wenn Sie die verwenden möchten DateFormatters Objekt in der Shell.