spark-sql-spark技术分享-第35页

Generator supports Java code generation (aka whole-stage codegen) conditionally, i.e. only when a physical operator is not marked as CodegenFallback.

Generator uses terminate to inform that there are no more rows to process, clean up code, and additional rows can be made here.



terminate(): TraversableOnce[InternalRow] = Nil

1

2

3

4

5

terminate(): TraversableOnce[InternalRow] = Nil

Name Description

CollectionGenerator

ExplodeBase

Explode

GeneratorOuter

HiveGenericUDTF

Inline

Corresponds to inline and inline_outer functions.

Represents an unresolved generator.

Created when AstBuilder creates Generate unary logical operator for LATERAL VIEW that corresponds to the following:



LATERAL VIEW (OUTER)?
generatorFunctionName (arg1, arg2, ...)
tblName
AS? col1, col2, ...

1

2

3

4

5

6

7

8

LATERAL VIEW (OUTER)?

generatorFunctionName (arg1, arg2, ...)

tblName

AS? col1, col2, ...

Note	`UnresolvedGenerator` is resolved to `Generator` by ResolveFunctions logical evaluation rule.

UserDefinedGenerator

Used exclusively in the deprecated explode operator

Note

You can only have one generator per select clause that is enforced by ExtractGenerator logical evaluation rule, e.g.



scala> xys.select(explode($"xs"), explode($"ys")).show
org.apache.spark.sql.AnalysisException: Only one generator allowed per select clause but found 2: explode(xs), explode(ys);
  at org.apache.spark.sql.catalyst.analysis.Analyzer$ExtractGenerator$$anonfun$apply$20.applyOrElse(Analyzer.scala:1670)
  at org.apache.spark.sql.catalyst.analysis.Analyzer$ExtractGenerator$$anonfun$apply$20.applyOrElse(Analyzer.scala:1662)
  at org.apache.spark.sql.catalyst.plans.logical.LogicalPlan$$anonfun$resolveOperators$1.apply(LogicalPlan.scala:62)

1

2

3

4

5

6

7

8

9

scala> xys.select(explode($"xs"), explode($"ys")).show

org.apache.spark.sql.AnalysisException: Only one generator allowed per select clause but found 2: explode(xs), explode(ys);

at org.apache.spark.sql.catalyst.analysis.Analyzer$ExtractGenerator$$anonfun$apply$20.applyOrElse(Analyzer.scala:1670)

at org.apache.spark.sql.catalyst.analysis.Analyzer$ExtractGenerator$$anonfun$apply$20.applyOrElse(Analyzer.scala:1662)

at org.apache.spark.sql.catalyst.plans.logical.LogicalPlan$$anonfun$resolveOperators$1.apply(LogicalPlan.scala:62)

If you want to have more than one generator in a structured query you should use LATERAL VIEW which is supported in SQL only, e.g.



val arrayTuple = (Array(1,2,3), Array("a","b","c"))
val ncs = Seq(arrayTuple).toDF("ns", "cs")

scala> ncs.show
+---------+---------+
|       ns|       cs|
+---------+---------+
|[1, 2, 3]|[a, b, c]|
+---------+---------+

scala> ncs.createOrReplaceTempView("ncs")

val q = """
  SELECT n, c FROM ncs
  LATERAL VIEW explode(ns) nsExpl AS n
  LATERAL VIEW explode(cs) csExpl AS c
"""

scala> sql(q).show
+---+---+
|  n|  c|
+---+---+
|  1|  a|
|  1|  b|
|  1|  c|
|  2|  a|
|  2|  b|
|  2|  c|
|  3|  a|
|  3|  b|
|  3|  c|
+---+---+

1

2

3

4

5

6

7

8

9

10

11

12

13

14

15

16

17

18

19

20

21

22

23

24

25

26

27

28

29

30

31

32

33

34

35

36

val arrayTuple = (Array(1,2,3), Array("a","b","c"))

val ncs = Seq(arrayTuple).toDF("ns", "cs")

scala> ncs.show

+---------+---------+

| ns| cs|

+---------+---------+

|[1, 2, 3]|[a, b, c]|

+---------+---------+

scala> ncs.createOrReplaceTempView("ncs")

val q = """

SELECT n, c FROM ncs

LATERAL VIEW explode(ns) nsExpl AS n

LATERAL VIEW explode(cs) csExpl AS c

"""

scala> sql(q).show

+---+---+

| n| c|

+---+---+

| 1| a|

| 1| b|

| 1| c|

| 2| a|

| 2| b|

| 2| c|

| 3| a|

| 3| b|

| 3| c|

+---+---+

Generator Contract



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

trait Generator extends Expression {
  // only required methods that have no implementation
  def elementSchema: StructType
  def eval(input: InternalRow): TraversableOnce[InternalRow]
}

1

2

3

4

5

6

7

8

9

10

11

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

trait Generator extends Expression {

// only required methods that have no implementation

def elementSchema: StructType

def eval(input: InternalRow): TraversableOnce[InternalRow]

}

Table 2. (Subset of) Generator Contract
Method	Description
`elementSchema`	Schema of the elements to be generated
`eval`

First

2012-08-22admin阅读(1499)

First Aggregate Function Expression

First is a DeclarativeAggregate function expression that is created when:

AstBuilder is requested to parse a FIRST statement
first standard function is used
first and first_value SQL functions are used



val sqlText = "FIRST (organizationName IGNORE NULLS)"
val e = spark.sessionState.sqlParser.parseExpression(sqlText)
scala> :type e
org.apache.spark.sql.catalyst.expressions.Expression

import org.apache.spark.sql.catalyst.expressions.aggregate.AggregateExpression
val aggExpr = e.asInstanceOf[AggregateExpression]

import org.apache.spark.sql.catalyst.expressions.aggregate.First
val f = aggExpr.aggregateFunction
scala> println(f.simpleString)
first('organizationName) ignore nulls

1

2

3

4

5

6

7

8

9

10

11

12

13

14

15

16

val sqlText = "FIRST (organizationName IGNORE NULLS)"

val e = spark.sessionState.sqlParser.parseExpression(sqlText)

scala> :type e

org.apache.spark.sql.catalyst.expressions.Expression

import org.apache.spark.sql.catalyst.expressions.aggregate.AggregateExpression

val aggExpr = e.asInstanceOf[AggregateExpression]

import org.apache.spark.sql.catalyst.expressions.aggregate.First

val f = aggExpr.aggregateFunction

scala> println(f.simpleString)

first('organizationName) ignore nulls

When requested to evaluate (and return the final value), First simply returns a AttributeReference (with first name and the data type of the child expression).

Tip	Use first operator from the Catalyst DSL to create an `First` aggregate function expression, e.g. for testing or Spark SQL internals exploration.

Catalyst DSL — `first` Operator



first(e: Expression): Expression

1

2

3

4

5

first(e: Expression): Expression

first creates a First expression and requests it to convert to a AggregateExpression.



import org.apache.spark.sql.catalyst.dsl.expressions._
val e = first('orgName)

scala> println(e.numberedTreeString)
00 first('orgName, false)
01 +- first('orgName)()
02    :- 'orgName
03    +- false

import org.apache.spark.sql.catalyst.expressions.aggregate.AggregateExpression
val aggExpr = e.asInstanceOf[AggregateExpression]

import org.apache.spark.sql.catalyst.expressions.aggregate.First
val f = aggExpr.aggregateFunction
scala> println(f.simpleString)
first('orgName)()

1

2

3

4

5

6

7

8

9

10

11

12

13

14

15

16

17

18

19

20

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

val e = first('orgName)

scala> println(e.numberedTreeString)

00 first('orgName, false)

01 +- first('orgName)()

02 :- 'orgName

03 +- false

import org.apache.spark.sql.catalyst.expressions.aggregate.AggregateExpression

val aggExpr = e.asInstanceOf[AggregateExpression]

import org.apache.spark.sql.catalyst.expressions.aggregate.First

val f = aggExpr.aggregateFunction

scala> println(f.simpleString)

first('orgName)()

Creating First Instance

First takes the following when created:

Child expression
ignoreNullsExpr flag expression

ExplodeBase Contract

2012-08-21admin阅读(1343)

ExplodeBase Base Generator Expression

ExplodeBase is the base class for Explode and PosExplode generator expressions.

ExplodeBase is a unary expression and Generator with CodegenFallback.

Explode Generator Unary Expression

Explode is a unary expression that produces a sequence of records for each value in the array or map.

Explode is a result of executing explode function (in SQL and functions)



scala> sql("SELECT explode(array(10,20))").explain
== Physical Plan ==
Generate explode([10,20]), false, false, [col#68]
+- Scan OneRowRelation[]

scala> sql("SELECT explode(array(10,20))").queryExecution.optimizedPlan.expressions(0)
res18: org.apache.spark.sql.catalyst.expressions.Expression = explode([10,20])

val arrayDF = Seq(Array(0,1)).toDF("array")
scala> arrayDF.withColumn("num", explode('array)).explain
== Physical Plan ==
Generate explode(array#93), true, false, [array#93, num#102]
+- LocalTableScan [array#93]

1

2

3

4

5

6

7

8

9

10

11

12

13

14

15

16

17

scala> sql("SELECT explode(array(10,20))").explain

== Physical Plan ==

Generate explode([10,20]), false, false, [col#68]

+- Scan OneRowRelation[]

scala> sql("SELECT explode(array(10,20))").queryExecution.optimizedPlan.expressions(0)

res18: org.apache.spark.sql.catalyst.expressions.Expression = explode([10,20])

val arrayDF = Seq(Array(0,1)).toDF("array")

scala> arrayDF.withColumn("num", explode('array)).explain

== Physical Plan ==

Generate explode(array#93), true, false, [array#93, num#102]

+- LocalTableScan [array#93]

PosExplode

Caution

FIXME

ExpectsInputTypes Contract

2012-08-20admin阅读(1253)

ExpectsInputTypes Contract

ExpectsInputTypes…FIXME

Exists

2012-08-19admin阅读(1488)

Exists — Correlated Predicate Subquery Expression

Exists is a SubqueryExpression and a predicate expression (i.e. the result data type is always boolean).

Exists is created when:

ResolveSubquery is requested to resolveSubQueries
PullupCorrelatedPredicates is requested to rewriteSubQueries
AstBuilder is requested to visitExists (in SQL statements)

Exists cannot be evaluated, i.e. produce a value given an internal row.

When requested to evaluate or doGenCode, Exists simply reports a UnsupportedOperationException.



Cannot evaluate expression: [this]

1

2

3

4

5

Cannot evaluate expression: [this]

Exists is never nullable.

Exists uses the following text representation:



exists#[exprId] [conditionString]

1

2

3

4

5

exists#[exprId] [conditionString]

When requested for a canonicalized version, Exists creates a new instance with…FIXME

Creating Exists Instance

Exists takes the following when created:

Subquery logical plan
Child expressions
ExprId

ExecSubqueryExpression

2012-08-18admin阅读(1461)

ExecSubqueryExpression Contract — Catalyst Expressions with SubqueryExec Physical Operators

ExecSubqueryExpression is the contract for Catalyst expressions that contain a physical plan with SubqueryExec physical operator (i.e. PlanExpression[SubqueryExec]).



package org.apache.spark.sql.execution

abstract class ExecSubqueryExpression extends PlanExpression[SubqueryExec] {
  def updateResult(): Unit
}

1

2

3

4

5

6

7

8

9

package org.apache.spark.sql.execution

abstract class ExecSubqueryExpression extends PlanExpression[SubqueryExec] {

def updateResult(): Unit

}

Table 1. ExecSubqueryExpression Contract
Method	Description
`updateResult`	Used exclusively when a physical operator is requested to waitForSubqueries (when executed as part of Physical Operator Execution Pipeline).

Table 2. ExecSubqueryExpressions
ExecSubqueryExpression	Description
InSubquery
ScalarSubquery

DeclarativeAggregate Contract — Unevaluable Aggregate Function Expressions

2012-08-17admin阅读(1601)

DeclarativeAggregate Contract — Unevaluable Aggregate Function Expressions

DeclarativeAggregate is an extension of the AggregateFunction Contract for aggregate function expressions that are unevaluable and use expressions for evaluation.

Note	An unevaluable expression cannot be evaluated to produce a value (neither in interpreted nor code-generated expression evaluations) and has to be resolved (replaced) to some other expressions or logical operators at analysis or optimization phases or they fail analysis.

Property Description

evaluateExpression



evaluateExpression: Expression

1

2

3

4

5

evaluateExpression: Expression

The expression that returns the final value for the aggregate function

Used when:

AggregationIterator is requested for the generateResultProjection
HashAggregateExec physical operator is requested to doProduceWithoutKeys and generateResultFunction
AggregateProcessor is created (when WindowExec physical operator is executed)

initialValues



initialValues: Seq[Expression]

1

2

3

4

5

initialValues: Seq[Expression]

mergeExpressions



mergeExpressions: Seq[Expression]

1

2

3

4

5

mergeExpressions: Seq[Expression]

updateExpressions



updateExpressions: Seq[Expression]

1

2

3

4

5

updateExpressions: Seq[Expression]

Table 2. DeclarativeAggregates (Direct Implementations)
DeclarativeAggregate	Description
AggregateWindowFunction	Contract for declarative window aggregate function expressions
Average
CentralMomentAgg
Corr
Count
Covariance
First
Last
Max
Min
SimpleTypedAggregateExpression
Sum

spark-sql 第35页

GetMapValue

GetMapValue

GetArrayItem

GetArrayItem

GetArrayStructFields

GetArrayStructFields

Generator

Generator Contract — Expressions to Generate Zero Or More Rows (aka Lateral Views)

Generator Contract

First

First Aggregate Function Expression

Catalyst DSL — `first` Operator

Creating First Instance

ExplodeBase Contract

ExplodeBase Base Generator Expression

Explode Generator Unary Expression

PosExplode

ExpectsInputTypes Contract

ExpectsInputTypes Contract

Exists

Exists — Correlated Predicate Subquery Expression

Creating Exists Instance

ExecSubqueryExpression

ExecSubqueryExpression Contract — Catalyst Expressions with SubqueryExec Physical Operators

DeclarativeAggregate Contract — Unevaluable Aggregate Function Expressions

DeclarativeAggregate Contract — Unevaluable Aggregate Function Expressions

欢迎关注：spark技术分享

关注公众号：spark技术分享

QQ咨询

回顶部

spark-sql 第35页

GetMapValue

GetArrayItem

GetArrayStructFields

Generator Contract — Expressions to Generate Zero Or More Rows (aka Lateral Views)

Generator Contract

First Aggregate Function Expression

Catalyst DSL — first Operator

Creating First Instance

ExplodeBase Base Generator Expression

Explode Generator Unary Expression

PosExplode

ExpectsInputTypes Contract

Exists — Correlated Predicate Subquery Expression

Creating Exists Instance

ExecSubqueryExpression Contract — Catalyst Expressions with SubqueryExec Physical Operators

DeclarativeAggregate Contract — Unevaluable Aggregate Function Expressions

欢迎关注：spark技术分享

关注公众号：spark技术分享

QQ咨询

回顶部

Catalyst DSL — `first` Operator