用户
 找回密码
 立即注册

QQ登录

只需一步,快速开始

搜索
查看: 1476|回复: 0

第八篇 Spark SQL Catalyst源码分析之UDF

[复制链接]

394

主题

412

帖子

2065

积分

管理员

Rank: 9Rank: 9Rank: 9

积分
2065

活跃会员热心会员推广达人宣传达人灌水之王突出贡献优秀版主荣誉管理论坛元老

发表于 2018-1-6 21:55:16 | 显示全部楼层 |阅读模式
----------------------------------------------------------
Spark SQL源码分析系列文章
在SQL的世界里,除了官方提供的常用的处理函数之外,一般都会提供可扩展的对外自定义函数接口,这已经成为一种事实的标准。
  在前面Spark SQL源码分析之核心流程一文中,已经介绍了Spark SQL Catalyst Analyzer的作用,其中包含了ResolveFunctions这个解析函数的功能。但是随着Spark1.1版本的发布,Spark SQL的代码有很多新完善和新功能了,和我先前基于1.0的源码分析多少有些不同,比如支持UDF:
  spark1.0及以前的实现:
[Java] 纯文本查看 复制代码
protected[sql] lazy val catalog: Catalog = new SimpleCatalog  
@transient  
protected[sql] lazy val analyzer: Analyzer =  
  new Analyzer(catalog, EmptyFunctionRegistry, caseSensitive = true) //EmptyFunctionRegistry空实现  
@transient  
protected[sql] val optimizer = Optimizer  


  Spark1.1及以后的实现:
[Java] 纯文本查看 复制代码
protected[sql] lazy val functionRegistry: FunctionRegistry = new SimpleFunctionRegistry //SimpleFunctionRegistry实现,支持简单的UDF  
  
@transient  
protected[sql] lazy val analyzer: Analyzer =  
  new Analyzer(catalog, functionRegistry, caseSensitive = true)  


一、引子:
  对于SQL语句中的函数,会经过SqlParser的的解析成UnresolvedFunction。UnresolvedFunction最后会被Analyzer解析。
SqlParser:
除了非官方定义的函数外,还可以定义自定义函数,sql parser会进行解析。
[Java] 纯文本查看 复制代码
ident ~ "(" ~ repsep(expression, ",") <~ ")" ^^ {  
    case udfName ~ _ ~ exprs => UnresolvedFunction(udfName, exprs)  

  将SqlParser传入的udfName和exprs封装成一个class class UnresolvedFunction继承自Expression。
  只是这个Expression的dataType等一系列属性和eval计算方法均无法访问,强制访问会抛出异常,因为它没有被Resolved,只是一个载体。
[Java] 纯文本查看 复制代码
case class UnresolvedFunction(name: String, children: Seq[Expression]) extends Expression {  
  override def dataType = throw new UnresolvedException(this, "dataType")  
  override def foldable = throw new UnresolvedException(this, "foldable")  
  override def nullable = throw new UnresolvedException(this, "nullable")  
  override lazy val resolved = false  
  
  // Unresolved functions are transient at compile time and don't get evaluated during execution.  
  override def eval(input: Row = null): EvaluatedType =  
    throw new TreeNodeException(this, s"No function to evaluate expression. type: ${this.nodeName}")  
  
  override def toString = s"'$name(${children.mkString(",")})"  
}<strong></strong>  


Analyzer:
  Analyzer初始化的时候会需要Catalog,database和table的元数据关系,以及FunctionRegistry来维护UDF名称和UDF实现的元数据,这里使用SimpleFunctionRegistry。
[Java] 纯文本查看 复制代码
/** 
 * Replaces [[UnresolvedFunction]]s with concrete [[catalyst.expressions.Expression Expressions]]. 
 */  
object ResolveFunctions extends Rule[LogicalPlan] {  
  def apply(plan: LogicalPlan): LogicalPlan = plan transform {  
    case q: LogicalPlan =>  
      q transformExpressions { //对当前LogicalPlan进行transformExpressions操作  
        case u @ UnresolvedFunction(name, children) if u.childrenResolved => //如果遍历到了UnresolvedFunction  
          registry.lookupFunction(name, children) //从UDF元数据表里查找udf函数  
      }  
  }  
}

二、UDF注册
2.1 UDFRegistration
   1.png
  registerFunction("len", (x:String)=>x.length)
  registerFunction是UDFRegistration下的方法,SQLContext现在实现了UDFRegistration这个trait,只要导入SQLContext,即可以使用udf功能。
  UDFRegistration核心方法registerFunction:
  registerFunction方法签名def registerFunction[T: TypeTag](name: String, func: Function1[_, T]): Unit
  接受一个udfName 和 一个FunctionN,可以是Function1 到Function22。即这个udf的参数只支持1-22个。(scala的痛啊)
  内部builder通过ScalaUdf来构造一个Expression,这里ScalaUdf继承自Expression(可以简单的理解目前的SimpleUDF即是一个Catalyst的一个Expression),传入scala的function作为UDF的实现,并且用反射检查字段类型是否是Catalyst允许的,见ScalaReflection.
[Java] 纯文本查看 复制代码
def registerFunction[T: TypeTag](name: String, func: Function1[_, T]): Unit = {  
def builder(e: Seq[Expression]) = ScalaUdf(func, ScalaReflection.schemaFor(typeTag[T]).dataType, e)//构造Expression  
functionRegistry.registerFunction(name, builder)//向SQLContext的functionRegistry(维护了一个hashMap来管理udf映射)注册  

2.2 注册Function:
注意:这里FunctionBuilder是一个type FunctionBuilder = Seq[Expression] => Expression
[Java] 纯文本查看 复制代码
class SimpleFunctionRegistry extends FunctionRegistry {  
  val functionBuilders = new mutable.HashMap[String, FunctionBuilder]() //udf映射关系维护[udfName,Expression]  
  
  def registerFunction(name: String, builder: FunctionBuilder) = { //put expression进Map  
    functionBuilders.put(name, builder)  
  }  
  
  override def lookupFunction(name: String, children: Seq[Expression]): Expression = {  
    functionBuilders(name)(children) //查找udf,返回Expression  
  }  
}  

至此,我们将一个scala function注册为一个catalyst的一个Expression,这就是spark的simple udf。
三、UDF计算:
UDF既然已经被封装为catalyst树里的一个Expression节点,那么计算的时候也就是计算ScalaUdf的eval方法。
先通过Row和表达式计算function所需要的参数,最后通过反射调用function,来达到计算udf的目的。
ScalaUdf继承自Expression:
2.png
scalaUdf接受一个function, dataType,和一系列表达式。
比较简单,看注释即可:
[Java] 纯文本查看 复制代码
case class ScalaUdf(function: AnyRef, dataType: DataType, children: Seq[Expression])  
  extends Expression {  
  
  type EvaluatedType = Any  
  
  def nullable = true  
  
  override def toString = s"scalaUDF(${children.mkString(",")})"  
 override def eval(input: Row): Any = {  
    val result = children.size match {  
      case 0 => function.asInstanceOf[() => Any]()  
      case 1 => function.asInstanceOf[(Any) => Any](children(0).eval(input)) //反射调用function  
      case 2 =>  
        function.asInstanceOf[(Any, Any) => Any](  
          children(0).eval(input), //表达式参数计算  
          children(1).eval(input))  
      case 3 =>  
        function.asInstanceOf[(Any, Any, Any) => Any](  
          children(0).eval(input),  
          children(1).eval(input),  
          children(2).eval(input))  
      case 4 =>  
     ......  
       case 22 => //scala function只支持22个参数,这里枚举了。  
        function.asInstanceOf[(Any, Any, Any, Any, Any, Any, Any, Any, Any, Any, Any, Any, Any, Any, Any, Any, Any, Any, Any, Any, Any, Any) => Any](  
          children(0).eval(input),  
          children(1).eval(input),  
          children(2).eval(input),  
          children(3).eval(input),  
          children(4).eval(input),  
          children(5).eval(input),  
          children(6).eval(input),  
          children(7).eval(input),  
          children(8).eval(input),  
          children(9).eval(input),  
          children(10).eval(input),  
          children(11).eval(input),  
          children(12).eval(input),  
          children(13).eval(input),  
          children(14).eval(input),  
          children(15).eval(input),  
          children(16).eval(input),  
          children(17).eval(input),  
          children(18).eval(input),  
          children(19).eval(input),  
          children(20).eval(input),  
          children(21).eval(input))  

四、总结
    Spark目前的UDF其实就是scala function。将scala function封装到一个Catalyst Expression当中,在进行sql计算时,使用同样的Eval方法对当前输入Row进行计算。
    编写一个spark udf非常简单,只需给UDF起个函数名,并且传递一个scala function即可。依靠scala函数编程的表现能力,使得编写scala udf比较简单,且相较hive的udf更容易使人理解。
——EOF——


原创文章,转载请注明:

回复

使用道具 举报

您需要登录后才可以回帖 登录 | 立即注册

本版积分规则

关闭

站长推荐 上一条 /4 下一条

返回顶部