Spark udf初始化

Question

我想在Spark SQL中创建一个基于regex的自定义UDF.我的偏好是创建一个内存驻留

 Map[String,Pattern]

其中Pattern指的是字符串键的已编译正则表达式版本.但要做到这一点,我们需要将地图创建放入UDF的"初始化"功能中.

那么Spark udf是否支持跨调用的持久状态(通过Spark SQL)？

请注意,HIVE确实支持UDF的生命周期.我使用它来生成Parse树作为初始化的一部分,因此UDF的实际调用是针对闪电般快速的树而不涉及解析.

Answer 1

让我们从导入和一些虚拟数据开始:

import org.apache.spark.sql.functions.udf
import scala.util.matching.Regex
import java.util.regex.Pattern

val df = sc.parallelize(Seq(
  ("foo", "this is bar"), ("foo", "this is foo"),
  ("bar", "foobar"), ("bar", "foo and foo")
)).toDF("type", "value")

和地图:

val patterns: Map[String, Pattern] = Seq(("foo", ".*foo.*"), ("bar", ".*bar.*"))
   .map{case (k, v) => (k, new Regex(v).pattern)}
   .toMap

现在我看到两种不同的选择:

• 制作patterns一个内部引用的广播变量udf

  val patternsBd = sc.broadcast(patterns)
  
  val typeMatchedViaBroadcast = udf((t: String, v: String) =>
    patternsBd.value.get(t).map(m => m.matcher(v).matches))
  
  df.withColumn("match", typeMatchedViaBroadcast($"type", $"value")).show
  
  // +----+-----------+-----+
  // |type|      value|match|
  // +----+-----------+-----+
  // | foo|this is bar|false|
  // | foo|this is foo| true|
  // | bar|     foobar| true|
  // | bar|foo and foo|false|
  // +----+-----------+-----+
  

• 在封闭内传递地图

  def makeTypeMatchedViaClosure(patterns: Map[String, Pattern]) = udf(
    (t: String, v: String) => patterns.get(t).map(m => m.matcher(v).matches))
  
  val typeMatchedViaClosure = makeTypeMatchedViaClosure(patterns)
  
  df.withColumn("match", typeMatchedViaClosure($"type", $"value")).show
  
  // +----+-----------+-----+
  // |type|      value|match|
  // +----+-----------+-----+
  // | foo|this is bar|false|
  // | foo|this is foo| true|
  // | bar|     foobar| true|
  // | bar|foo and foo|false|
  // +----+-----------+-----+