写点什么

大数据培训 spark SQL 中 count(*) 和 count(1) 源码分析

作者:@零度
  • 2022 年 4 月 11 日
  • 本文字数:1831 字

    阅读完需:约 6 分钟

​以下文章来源于数据仓库践行者


本文基于 spark 3.2


先给结论,在 spark sql 中 count(*)不管在运行效率方面,还是在最终展示结果方面 都等同于 count(1)。


案例展示


sql:


SELECT A,COUNT(*) AS CNT FROM TESTDATA2 GROUP BY A



从上面打印可以看到,count(*)在生成未解析的逻辑执行计划时就被转换成了 count(1)。


也就是说,在 spark sql 中 count(*)==count(1)。


源码分析


从 sql 生成的 AstTree 上来看,处理函数的节点是 functioncall:



在 AstBuilder 类中找到 visitFunctionCall 方法:



spark sql count 函数详解


count 函数的代码虽然简单,但是开发该代码的同学还是做了一些优化在里面_大数据培训


count 函数功能:


1、count(*) 返回检索到的行的总数,包括包含 null 的行。


2、count(expr[, expr...]) 返回提供的表达式均为非空的行数。


3、count(DISTINCT expr[, expr...]) 返回提供的表达式唯一且非空的行数。


一个点:


判断 expression 是不是为 null,先从 expression 的 nullable 的属性里拿,如果 nullable 的属性不能判断了,再用 isnull 来判断具体的值是不是为 null。


这样在一定程度上能节省计算资源。比如 count(1)这样的常量,1 一定是不为 null 的,属性里就可以确定了,不用再走一篇 eval 取值。


package org.apache.spark.sql.catalyst.expressions.aggregate


import org.apache.spark.sql.catalyst.analysis.TypeCheckResult


import org.apache.spark.sql.catalyst.dsl.expressions._


import org.apache.spark.sql.catalyst.expressions._


import org.apache.spark.sql.catalyst.trees.TreePattern.{COUNT, TreePattern}


import org.apache.spark.sql.internal.SQLConf


import org.apache.spark.sql.types._


// scalastyle:off line.size.limit


@ExpressionDescription(


usage = """


FUNC(*) - Returns the total number of retrieved rows, including rows containing null.


FUNC(expr[, expr...]) - Returns the number of rows for which the supplied expression(s) are all non-null.


FUNC(DISTINCT expr[, expr...]) - Returns the number of rows for which the supplied expression(s) are unique and non-null.


""",


examples = """


Examples:


SELECT FUNC(*) FROM VALUES (NULL), (5), (5), (20) AS tab(col);


4


SELECT FUNC(col) FROM VALUES (NULL), (5), (5), (20) AS tab(col);


3


SELECT FUNC(DISTINCT col) FROM VALUES (NULL), (5), (5), (10) AS tab(col);


2


""",


group = "agg_funcs",


since = "1.0.0")


// scalastyle:on line.size.limit


case class Count(children: Seq[Expression]) extends DeclarativeAggregate {


override def nullable: Boolean = false


final override val nodePatterns: Seq[TreePattern] = Seq(COUNT)


// Return data type.


override def dataType: DataType = LongType


override def checkInputDataTypes(): TypeCheckResult = {


if (children.isEmpty && !SQLConf.get.getConf(SQLConf.ALLOW_PARAMETERLESS_COUNT)) {


TypeCheckResult.TypeCheckFailure(s"$prettyName requires at least one argument. " +


s"If you have to call the function $prettyName without arguments, set the legacy " +


s"configuration ${SQLConf.ALLOW_PARAMETERLESS_COUNT.key} as true")


} else {


TypeCheckResult.TypeCheckSuccess


}


}


protected lazy val count = AttributeReference("count", LongType, nullable = false)()


override lazy val aggBufferAttributes = count :: Nil


override lazy val initialValues = Seq(


/* count = */ Literal(0L)


)


override lazy val mergeExpressions = Seq(


/* count = */ count.left + count.right


)


override lazy val evaluateExpression = count


override def defaultResult: Option[Literal] = Option(Literal(0L))


override lazy val updateExpressions = {


//去掉为空的 expression(这一层的判断是走 expression 里的 nullable 的属性)


val nullableChildren = children.filter(_.nullable)


if (nullableChildren.isEmpty) {


Seq(


/* count = */ count + 1L


)


} else {


// expression 的 nullable 的属性只能依赖拿到具体值判断时,用 isnull 方法来判断是不是空值


Seq(


/* count = */ If(nullableChildren.map(IsNull).reduce(Or), count, count + 1L)


)


}


}


override protected def withNewChildrenInternal(newChildren: IndexedSeq[Expression]): Count =


copy(children = newChildren)


}


object Count {


def apply(child: Expression): Count = Count(child :: Nil)


}


用户头像

@零度

关注

关注尚硅谷,轻松学IT 2021.11.23 加入

IT培训 www.atguigu.com

评论

发布
暂无评论
大数据培训spark SQL中count(*)和count(1)源码分析_大数据开发_@零度_InfoQ写作平台