Scala und Spark-UDF-Funktion

Habe ich ein einfaches UDF zu konvertieren oder zu extrahieren einige Werte aus einem Feld in ein temptabl in spark. Melde ich die Funktion aber wenn ich den Aufruf der Funktion mit sql wirft es eine NullPointerException. Unten ist meine Funktion und Ausführung. Ich bin mit dem Zeppelin. Pikanterweise lag dieser arbeiten war gestern, aber es gestoppt arbeiten an diesem morgen.

Funktion

def convert( time:String ) : String = {
  val sdf = new java.text.SimpleDateFormat("HH:mm")
  val time1 = sdf.parse(time)
  return sdf.format(time1)
}

Registrieren der Funktion

sqlContext.udf.register("convert",convert _)

Testen Sie die Funktion ohne SQL -- Dies funktioniert

convert(12:12:12) -> returns 12:12

Testen Sie die Funktion mit SQL im Zeppelin dieser AUSFÄLLT. 

%sql
select convert(time) from temptable limit 10

Struktur der temptable

root
 |-- date: string (nullable = true)
 |-- time: string (nullable = true)
 |-- serverip: string (nullable = true)
 |-- request: string (nullable = true)
 |-- resource: string (nullable = true)
 |-- protocol: integer (nullable = true)
 |-- sourceip: string (nullable = true)

Teil der stacktrace, den ich erhalte. 

java.lang.NullPointerException
    at org.apache.hadoop.hive.ql.exec.FunctionRegistry.getFunctionInfo(FunctionRegistry.java:643)
    at org.apache.hadoop.hive.ql.exec.FunctionRegistry.getFunctionInfo(FunctionRegistry.java:652)
    at org.apache.spark.sql.hive.HiveFunctionRegistry.lookupFunction(hiveUdfs.scala:54)
    at org.apache.spark.sql.hive.HiveContext$$anon$3.org$apache$spark$sql$catalyst$analysis$OverrideFunctionRegistry$$super$lookupFunction(HiveContext.scala:376)
    at org.apache.spark.sql.catalyst.analysis.OverrideFunctionRegistry$$anonfun$lookupFunction$2.apply(FunctionRegistry.scala:44)
    at org.apache.spark.sql.catalyst.analysis.OverrideFunctionRegistry$$anonfun$lookupFunction$2.apply(FunctionRegistry.scala:44)
    at scala.Option.getOrElse(Option.scala:120)
    at org.apache.spark.sql.catalyst.analysis.OverrideFunctionRegistry$class.lookupFunction(FunctionRegistry.scala:44)
InformationsquelleAutor fanbondi | 2016-07-28
  1. 13

    Udf verwenden, anstatt eine Funktion definieren, die direkt

    import org.apache.spark.sql.functions._

val convert = udf[String, String](time => {
        val sdf = new java.text.SimpleDateFormat("HH:mm")
        val time1 = sdf.parse(time)
        sdf.format(time1)
    }
)

    Einer udf-input-parameter ist die Spalte(oder Spalten). Und der return-type-Spalte. 

    case class UserDefinedFunction protected[sql] (
    f: AnyRef,
    dataType: DataType,
    inputTypes: Option[Seq[DataType]]) {

  def apply(exprs: Column*): Column = {
    Column(ScalaUDF(f, dataType, exprs.map(_.expr), inputTypes.getOrElse(Nil)))
  }
}
    InformationsquelleAutor Rockie Yang
  2. 2

    Haben Sie zum definieren Ihrer Funktion als UDF.

    import org.apache.spark.sql.expressions.UserDefinedFunction
import org.apache.spark.sql.functions.udf

val convertUDF: UserDefinedFunction = udf((time:String) => {
  val sdf = new java.text.SimpleDateFormat("HH:mm")
  val time1 = sdf.parse(time)
  sdf.format(time1)
})

    Neben Sie gelten würde, die Ihre UDF auf Ihre DataFrame.

    //assuming your DataFrame is already defined
dataFrame.withColumn("time", convertUDF(col("time"))) //using the same name replaces existing

    Nun zu deinem eigentlichen problem, ein Grund, warum Sie diesen Fehler empfangen werden konnten, weil Ihre DataFrame enthält Zeilen, die null-Werte. Wenn Sie die filter aus, bevor Sie die UDF, Sie sollten in der Lage sein, um fortzufahren kein problem.

    dataFrame.filter(col("time").isNotNull)

    Bin ich gespannt, was sonst bewirkt, dass eine NullPointerException, wenn ein UDF-anders als es die Begegnung mit einer null, wenn Sie finden einen Grund, anders als mein Vorschlag, ich würde froh sein zu wissen.

    InformationsquelleAutor kfkhalili
Schreibe einen Kommentar