webentwicklung-frage-antwort-db.com.de

SparkSQL: Wie gehe ich mit Nullwerten in benutzerdefinierten Funktionen um?

Gegebene Tabelle 1 mit einer Spalte "x" vom Typ String. Ich möchte Tabelle 2 mit einer Spalte "y" erstellen, die eine ganzzahlige Darstellung der in "x" angegebenen Datumszeichenfolgen ist.

Essential soll null Werte in Spalte "y" behalten.

Tabelle 1 (Datenrahmen df1):

+----------+
|         x|
+----------+
|2015-09-12|
|2015-09-13|
|      null|
|      null|
+----------+
root
 |-- x: string (nullable = true)

Tabelle 2 (Datenrahmen df2):

+----------+--------+                                                                  
|         x|       y|
+----------+--------+
|      null|    null|
|      null|    null|
|2015-09-12|20150912|
|2015-09-13|20150913|
+----------+--------+
root
 |-- x: string (nullable = true)
 |-- y: integer (nullable = true)

Während die benutzerdefinierte Funktion (udf) zum Konvertieren von Werten aus der Spalte "x" in die Werte der Spalte "y" wie folgt lautet:

val extractDateAsInt = udf[Int, String] (
  (d:String) => d.substring(0, 10)
      .filterNot( "-".toSet)
      .toInt )

und funktioniert, ist der Umgang mit Nullwerten nicht möglich.

Auch wenn ich sowas machen kann

val extractDateAsIntWithNull = udf[Int, String] (
  (d:String) => 
    if (d != null) d.substring(0, 10).filterNot( "-".toSet).toInt 
    else 1 )

Ich habe keine Möglichkeit gefunden, null Werte über udfs "zu erzeugen" (da Ints natürlich nicht null sein kann).

Meine aktuelle Lösung zur Erstellung von df2 (Tabelle 2) lautet wie folgt:

// holds data of table 1  
val df1 = ... 

// filter entries from df1, that are not null
val dfNotNulls = df1.filter(df1("x")
  .isNotNull)
  .withColumn("y", extractDateAsInt(df1("x")))
  .withColumnRenamed("x", "right_x")

// create df2 via a left join on df1 and dfNotNull having 
val df2 = df1.join( dfNotNulls, df1("x") === dfNotNulls("right_x"), "leftouter" ).drop("right_x")

Fragen:

  • Die derzeitige Lösung scheint umständlich (und wahrscheinlich nicht effizient in Bezug auf die Leistung) zu sein. Gibt es einen besseren Weg?
  • @ Spark-Entwickler: Gibt es einen Typ NullableInt geplant/verfügbar, so dass das folgende udf möglich ist (siehe Code-Auszug)?

Code-Auszug

val extractDateAsNullableInt = udf[NullableInt, String] (
  (d:String) => 
    if (d != null) d.substring(0, 10).filterNot( "-".toSet).toInt 
    else null )
scalaapache-sparkapache-spark-sqluser-defined-functionsnullable
27
Martin Senne

Hier ist Option nützlich:

val extractDateAsOptionInt = udf((d: String) => d match {
  case null => None
  case s => Some(s.substring(0, 10).filterNot("-".toSet).toInt)
})

oder um es im Allgemeinen etwas sicherer zu machen:

import scala.util.Try

val extractDateAsOptionInt = udf((d: String) => Try(
  d.substring(0, 10).filterNot("-".toSet).toInt
).toOption)

Alle Verdienste gehen an Dmitriy Selivanov , die auf diese Lösung als (fehlende?) Bearbeitung hingewiesen haben hier .

Alternativ kann null auch außerhalb der UDF behandelt werden:

import org.Apache.spark.sql.functions.{lit, when}
import org.Apache.spark.sql.types.IntegerType

val extractDateAsInt = udf(
   (d: String) => d.substring(0, 10).filterNot("-".toSet).toInt
)

df.withColumn("y",
  when($"x".isNull, lit(null))
    .otherwise(extractDateAsInt($"x"))
    .cast(IntegerType)
)
51
zero323

Scala hat tatsächlich eine nette Factory-Funktion, Option (), die dies noch prägnanter machen kann:

val extractDateAsOptionInt = udf((d: String) => 
  Option(d).map(_.substring(0, 10).filterNot("-".toSet).toInt))

Intern führt die Apply-Methode des Option-Objekts nur die Nullprüfung für Sie durch:

def apply[A](x: A): Option[A] = if (x == null) None else Some(x)
11
tristanbuckner

Ergänzungscode

Mit der Nice Antwort von @ zero323 habe ich den folgenden Code erstellt, um benutzerdefinierte Funktionen zur Verfügung zu haben, die mit Nullwerten wie beschrieben umgehen. Hoffe, es ist hilfreich für andere!

/**
 * Set of methods to construct [[org.Apache.spark.sql.UserDefinedFunction]]s that
 * handle `null` values.
 */
object NullableFunctions {

  import org.Apache.spark.sql.functions._
  import scala.reflect.runtime.universe.{TypeTag}
  import org.Apache.spark.sql.UserDefinedFunction

  /**
   * Given a function A1 => RT, create a [[org.Apache.spark.sql.UserDefinedFunction]] such that
   *   * if fnc input is null, None is returned. This will create a null value in the output Spark column.
   *   * if A1 is non null, Some( f(input) will be returned, thus creating f(input) as value in the output column.
   * @param f function from A1 => RT
   * @tparam RT return type
   * @tparam A1 input parameter type
   * @return a [[org.Apache.spark.sql.UserDefinedFunction]] with the behaviour describe above
   */
  def nullableUdf[RT: TypeTag, A1: TypeTag](f: Function1[A1, RT]): UserDefinedFunction = {
    udf[Option[RT],A1]( (i: A1) => i match {
      case null => None
      case s => Some(f(i))
    })
  }

  /**
   * Given a function A1, A2 => RT, create a [[org.Apache.spark.sql.UserDefinedFunction]] such that
   *   * if on of the function input parameters is null, None is returned.
   *     This will create a null value in the output Spark column.
   *   * if both input parameters are non null, Some( f(input) will be returned, thus creating f(input1, input2)
   *     as value in the output column.
   * @param f function from A1 => RT
   * @tparam RT return type
   * @tparam A1 input parameter type
   * @tparam A2 input parameter type
   * @return a [[org.Apache.spark.sql.UserDefinedFunction]] with the behaviour describe above
   */
  def nullableUdf[RT: TypeTag, A1: TypeTag, A2: TypeTag](f: Function2[A1, A2, RT]): UserDefinedFunction = {
    udf[Option[RT], A1, A2]( (i1: A1, i2: A2) =>  (i1, i2) match {
      case (null, _) => None
      case (_, null) => None
      case (s1, s2) => Some((f(s1,s2)))
    } )
  }
}
10
Martin Senne