Comment puis-je passer des paramètres supplémentaires pour Udf Spark SQL?

Je veux analyser les colonnes de date dans un DataFrame, et pour chaque colonne de la date de la résolution pour la date peut changer (c'est à dire 2011/01/10 => 2011 /01 si la résolution est réglée sur "Mois").

J'ai écrit le code suivant:

def convertDataFrame(dataframe: DataFrame, schema : Array[FieldDataType], resolution: Array[DateResolutionType]) : DataFrame =
{
  import org.apache.spark.sql.functions._
  val convertDateFunc = udf{(x:String, resolution: DateResolutionType) => SparkDateTimeConverter.convertDate(x, resolution)}
  val convertDateTimeFunc = udf{(x:String, resolution: DateResolutionType) => SparkDateTimeConverter.convertDateTime(x, resolution)}

  val allColNames = dataframe.columns
  val allCols = allColNames.map(name => dataframe.col(name))

  val mappedCols =
  {
    for(i <- allCols.indices) yield
    {
      schema(i) match
      {
        case FieldDataType.Date => convertDateFunc(allCols(i), resolution(i)))
        case FieldDataType.DateTime => convertDateTimeFunc(allCols(i), resolution(i))
        case _ => allCols(i)
      }
    }
  }

  dataframe.select(mappedCols:_*)

}}

Toutefois, il ne fonctionne pas. Il semble que je ne peux passer Columns à l'Udf. Et je me demande si il sera très lente si je convertir le DataFrame à RDD et d'appliquer la fonction sur chaque ligne.

Quelqu'un sait-il la bonne solution? Merci!!!!

InformationsquelleAutor DarkZero | 2016-02-22
  1. 34

    Il suffit d'utiliser un peu de nourrissage:

    def convertDateFunc(resolution: DateResolutionType) = udf((x:String) => 
  SparkDateTimeConverter.convertDate(x, resolution))

    et de l'utiliser comme suit:

    case FieldDataType.Date => convertDateFunc(resolution(i))(allCols(i))

    Sur une note de côté, vous devriez jeter un oeil à sql.functions.trunc et sql.functions.date_format. Ceux-ci devraient au moins une partie de la tâche sans l'aide de fonctions définies par l'utilisateur à tous.

    Note:

    Spark 2.2 ou ultérieure, vous pouvez utiliser typedLit fonction:

    import org.apache.spark.sql.functions.typedLit

    qui prennent en charge un large éventail de littéraux comme Seq ou Map.

    • Je vous remercie pour votre réponse et à l'intuition de nourrissage!
    • J'ai écrit un tutoriel sur la façon d'utiliser nourrissage pour créer l'Étincelle de l'UDF, qui accepte des paramètres supplémentaires à l'invocation de temps. gist.github.com/andrearota/5910b5c5ac65845f23856b2415474c38
    InformationsquelleAutor zero323
  2. 14

    Vous pouvez créer un littéral Column pour passer d'une fonction à l'aide de la lit(...) fonction définie dans org.apache.spark.sql.functions

    Par exemple:

    val takeRight = udf((s: String, i: Int) => s.takeRight(i))
df.select(takeRight($"stringCol", lit(1)))
    • Merci, j'ai d'abord utilisé lit ainsi, mais il s'avère que son rendement n'est pas aussi bonne que l'autre réponse...
    InformationsquelleAutor Michael Armbrust