scala用户定义函数在spark sql中不起作用

Posted 2023-04-18

【中文标题】scala用户定义函数在spark sql中不起作用

【英文标题】：scala user define function not working in sparksql

【发布时间】：2017-08-12 13:22:47

【问题描述】：

我写了一个 UDF，它基本上计算给定的 IP 地址是否在 cidr 列表中。我可以从 scala 调用我的 UDF，它工作正常，但是当我从 spark sql 调用 udf 时，它抛出了这个错误。请帮帮我。

%spark
def isinlist = (ip:String) => 
  import org.apache.commons.net.util.SubnetUtils

  def checkipinrange = (cidr:String,ip:String) => 
    val utils = new SubnetUtils(cidr);
    val isInRange = utils.getInfo().isInRange(ip);
    if (isInRange) 
      true
     else 
      false
    
  
  sqlContext.udf.register("checkipinrange",checkipinrange)

  val query=s"""select * 
                from tag_ip  
                where checkipinrange(tag_ip.cidr, '$ip') """
  val validrange = sqlContext.sql(query)
  if(validrange.count > 0) 
    true
   else 
    false
  

isinlist("5.9.29.73")
sqlContext.udf.register("isinlist",isinlist)

tag_ip 是一个 cidr ip 范围列表。这里 isinlist 函数工作正常。但是当我从 spark sql 调用 isinlist 函数时，它会在下面显示错误。

java.lang.NullPointerException
at $line926276415525.$read$$iwC$$iwC$$iwC$$iwC$$iwC$$$$3baf9f919752f0ab1f5a31ad94af9f4$$$$$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$anonfun$isinlist$1.apply(<console>:198)
at $line926276415525.$read$$iwC$$iwC$$iwC$$iwC$$iwC$$$$3baf9f919752f0ab1f5a31ad94af9f4$$$$$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$anonfun$isinlist$1.apply(<console>:184)
at org.apache.spark.sql.catalyst.expressions.GeneratedClass$SpecificUnsafeProjection.apply(Unknown Source)
at org.apache.spark.sql.execution.Project$$anonfun$1$$anonfun$apply$1.apply(basicOperators.scala:51)
at org.apache.spark.sql.execution.Project$$anonfun$1$$anonfun$apply$1.apply(basicOperators.scala:49)
at scala.collection.Iterator$$anon$11.next(Iterator.scala:328)
at scala.collection.Iterator$$anon$11.next(Iterator.scala:328)
at scala.collection.Iterator$$anon$11.next(Iterator.scala:328)
at scala.collection.Iterator$$anon$10.next(Iterator.scala:312)
at scala.collection.Iterator$class.foreach(Iterator.scala:727)
at scala.collection.AbstractIterator.foreach(Iterator.scala:1157)
at scala.collection.generic.Growable$class.$plus$plus$eq(Growable.scala:48)
at scala.collection.mutable.ArrayBuffer.$plus$plus$eq(ArrayBuffer.scala:103)
at scala.collection.mutable.ArrayBuffer.$plus$plus$eq(ArrayBuffer.scala:47)
at scala.collection.TraversableOnce$class.to(TraversableOnce.scala:273)
at scala.collection.AbstractIterator.to(Iterator.scala:1157)
at scala.collection.TraversableOnce$class.toBuffer(TraversableOnce.scala:265)
at scala.collection.AbstractIterator.toBuffer(Iterator.scala:1157)
at scala.collection.TraversableOnce$class.toArray(TraversableOnce.scala:252)
at scala.collection.AbstractIterator.toArray(Iterator.scala:1157)
at org.apache.spark.sql.execution.SparkPlan$$anonfun$5.apply(SparkPlan.scala:212)
at org.apache.spark.sql.execution.SparkPlan$$anonfun$5.apply(SparkPlan.scala:212)
at org.apache.spark.SparkContext$$anonfun$runJob$5.apply(SparkContext.scala:1858)
at org.apache.spark.SparkContext$$anonfun$runJob$5.apply(SparkContext.scala:1858)
at org.apache.spark.scheduler.ResultTask.runTask(ResultTask.scala:66)

有人可以帮我解决问题吗？

【问题讨论】：

上面的代码有效吗？你是如何从 sql 查询中调用 isinlist 的？

【参考方案1】：

您应该检查null 值。例如：

val isInRange = ip != null && utils.getInfo().isInRange(ip);