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

RunnableCommand Contract — Generic Logical Command with Side Effects

2012-10-15admin阅读(1454)

RunnableCommand Contract — Generic Logical Command with Side Effects

RunnableCommand is the generic logical command that is executed eagerly for its side effects.

RunnableCommand defines one abstract method run that computes a collection of Row records with the side effect, i.e. the result of executing a command.



run(sparkSession: SparkSession): Seq[Row]

1

2

3

4

5

run(sparkSession: SparkSession): Seq[Row]

Note	`RunnableCommand` logical operator is resolved to ExecutedCommandExec physical operator in BasicOperators execution planning strategy.

Note	`run` is executed when: `ExecutedCommandExec` executes logical RunnableCommand and caches the result as InternalRows `InsertIntoHadoopFsRelationCommand` is executed `QueryExecution` is requested to transform the result of executing DescribeTableCommand to a Hive-compatible output format

RunnableCommand Description

AddFileCommand

AddJarCommand

AlterDatabasePropertiesCommand

AlterTableAddPartitionCommand

AlterTableChangeColumnCommand

AlterTableDropPartitionCommand

AlterTableRecoverPartitionsCommand

AlterTableRenameCommand

AlterTableRenamePartitionCommand

AlterTableSerDePropertiesCommand

AlterTableSetLocationCommand

AlterTableSetPropertiesCommand

AlterTableUnsetPropertiesCommand

AlterViewAsCommand

AnalyzeColumnCommand

AnalyzePartitionCommand

AnalyzeTableCommand

CacheTableCommand

When executed, CacheTableCommand creates a DataFrame followed by registering a temporary view for the optional query.



CACHE LAZY? TABLE [table] (AS? [query])?

1

2

3

4

5

CACHE LAZY? TABLE [table] (AS? [query])?

CacheTableCommand requests the session-specific Catalog to cache the table.

Note	`CacheTableCommand` uses `SparkSession` to access the `Catalog`.

If the caching is not LAZY (which is not by default), CacheTableCommand creates a DataFrame for the table and counts the rows (that will trigger the caching).

Note	`CacheTableCommand` uses a Spark SQL pattern to trigger DataFrame caching by executing `count` operation.



val q = "CACHE TABLE ids AS SELECT * from range(5)"
scala> println(sql(q).queryExecution.logical.numberedTreeString)
00 CacheTableCommand `ids`, false
01    +- 'Project [*]
02       +- 'UnresolvedTableValuedFunction range, [5]

val q2 = "CACHE LAZY TABLE ids"
scala> println(sql(q2).queryExecution.logical.numberedTreeString)
17/05/17 06:16:39 WARN CacheManager: Asked to cache already cached data.
00 CacheTableCommand `ids`, true

1

2

3

4

5

6

7

8

9

10

11

12

13

14

val q = "CACHE TABLE ids AS SELECT * from range(5)"

scala> println(sql(q).queryExecution.logical.numberedTreeString)

00 CacheTableCommand `ids`, false

01 +- 'Project [*]

02 +- 'UnresolvedTableValuedFunction range, [5]

val q2 = "CACHE LAZY TABLE ids"

scala> println(sql(q2).queryExecution.logical.numberedTreeString)

17/05/17 06:16:39 WARN CacheManager: Asked to cache already cached data.

00 CacheTableCommand `ids`, true

ClearCacheCommand

CreateDatabaseCommand

CreateDataSourceTableAsSelectCommand

When executed, …FIXME

Used exclusively when DataSourceAnalysis posthoc logical resolution rule resolves a CreateTable logical operator with queries using non-Hive table providers (which is when DataFrameWriter saves a DataFrame to a non-Hive table or for Create Table As Select SQL statements)

CreateDataSourceTableCommand

CreateFunctionCommand

CreateHiveTableAsSelectCommand

CreateTableCommand

CreateTableLikeCommand

CreateTempViewUsing

CreateViewCommand

DescribeColumnCommand

DescribeDatabaseCommand

DescribeFunctionCommand

DescribeTableCommand

DropDatabaseCommand

DropFunctionCommand

DropTableCommand

ExplainCommand

InsertIntoDataSourceCommand

InsertIntoHadoopFsRelationCommand

InsertIntoHiveTable

ListFilesCommand

ListJarsCommand

LoadDataCommand

RefreshResource

RefreshTable

ResetCommand

SaveIntoDataSourceCommand

When executed, requests DataSource to write a DataFrame to a data source per save mode.

Used exclusively when DataFrameWriter is requested to save a DataFrame to a data source.

SetCommand

SetDatabaseCommand

ShowColumnsCommand

ShowCreateTableCommand

ShowDatabasesCommand

ShowFunctionsCommand

ShowPartitionsCommand

ShowTablePropertiesCommand

ShowTablesCommand

StreamingExplainCommand

TruncateTableCommand

UncacheTableCommand

Command Contract — Eagerly-Executed Logical Operator

2012-10-14admin阅读(1485)

Command Contract — Eagerly-Executed Logical Operator

Command is the marker interface for logical operators that represent non-query commands that are executed early in the query plan lifecycle (unlike logical plans in general).

Note	`Command` is executed when a `Dataset` is requested for the logical plan (which is after the query has been analyzed).

Command has no output schema by default.

Command has no child logical operators (which makes it similar to leaf logical operators).

Table 1. Commands (Direct Implementations)
Command	Description
DataWritingCommand
RunnableCommand

LogicalPlan Contract — Logical Operator with Children and Expressions / Logical Query Plan

2012-10-13admin阅读(1828)

LogicalPlan Contract — Logical Relational Operator with Children and Expressions / Logical Query Plan

LogicalPlan is an extension of the QueryPlan contract for logical operators to build a logical query plan (i.e. a tree of logical operators).

Note	A logical query plan is a tree of nodes of logical operators that in turn can have (trees of) Catalyst expressions. In other words, there are at least two trees at every level (operator).

LogicalPlan can be resolved.

In order to get the logical plan of a structured query you should use the QueryExecution.



scala> :type q
org.apache.spark.sql.Dataset[Long]

val plan = q.queryExecution.logical
scala> :type plan
org.apache.spark.sql.catalyst.plans.logical.LogicalPlan

1

2

3

4

5

6

7

8

9

10

scala> :type q

org.apache.spark.sql.Dataset[Long]

val plan = q.queryExecution.logical

scala> :type plan

org.apache.spark.sql.catalyst.plans.logical.LogicalPlan

LogicalPlan goes through execution stages (as a QueryExecution). In order to convert a LogicalPlan to a QueryExecution you should use SessionState and request it to “execute” the plan.



scala> :type spark
org.apache.spark.sql.SparkSession

// You could use Catalyst DSL to create a logical query plan
scala> :type plan
org.apache.spark.sql.catalyst.plans.logical.LogicalPlan

val qe = spark.sessionState.executePlan(plan)
scala> :type qe
org.apache.spark.sql.execution.QueryExecution

1

2

3

4

5

6

7

8

9

10

11

12

13

14

scala> :type spark

org.apache.spark.sql.SparkSession

// You could use Catalyst DSL to create a logical query plan

scala> :type plan

org.apache.spark.sql.catalyst.plans.logical.LogicalPlan

val qe = spark.sessionState.executePlan(plan)

scala> :type qe

org.apache.spark.sql.execution.QueryExecution

Note

A common idiom in Spark SQL to make sure that a logical plan can be analyzed is to request a SparkSession for the SessionState that is in turn requested to execute the logical plan (which simply creates a QueryExecution).



scala> :type plan
org.apache.spark.sql.catalyst.plans.logical.LogicalPlan

val qe = sparkSession.sessionState.executePlan(plan)
qe.assertAnalyzed()
// the following gives the analyzed logical plan
// no exceptions are expected since analysis went fine
val analyzedPlan = qe.analyzed

1

2

3

4

5

6

7

8

9

10

11

12

scala> :type plan

org.apache.spark.sql.catalyst.plans.logical.LogicalPlan

val qe = sparkSession.sessionState.executePlan(plan)

qe.assertAnalyzed()

// the following gives the analyzed logical plan

// no exceptions are expected since analysis went fine

val analyzedPlan = qe.analyzed

Note	Another common idiom in Spark SQL to convert a `LogicalPlan` into a `Dataset` is to use Dataset.ofRows internal method that executes the logical plan followed by creating a Dataset with the QueryExecution and a RowEncoder.

A logical operator is considered partially resolved when its child operators are resolved (aka children resolved).

A logical operator is (fully) resolved to a specific schema when all expressions and the children are resolved.



scala> plan.resolved
res2: Boolean = true

1

2

3

4

5

6

scala> plan.resolved

res2: Boolean = true

A logical plan knows the size of objects that are results of query operators, like join, through Statistics object.



scala> val stats = plan.statistics
stats: org.apache.spark.sql.catalyst.plans.logical.Statistics = Statistics(8,false)

1

2

3

4

5

6

scala> val stats = plan.statistics

stats: org.apache.spark.sql.catalyst.plans.logical.Statistics = Statistics(8,false)

A logical plan knows the maximum number of records it can compute.



scala> val maxRows = plan.maxRows
maxRows: Option[Long] = None

1

2

3

4

5

6

scala> val maxRows = plan.maxRows

maxRows: Option[Long] = None

LogicalPlan can be streaming if it contains one or more structured streaming sources.

Note	`LogicalPlan` is in the end transformed to a physical query plan.

Table 1. Logical Operators / Specialized Logical Plans
LogicalPlan	Description
LeafNode	Logical operator with no child operators
`UnaryNode`	Logical plan with a single child logical operator
`BinaryNode`	Logical operator with two child logical operators
Command
RunnableCommand

Table 2. LogicalPlan’s Internal Registries and Counters
Name	Description
`statsCache`	Cached plan statistics (as `Statistics`) of the `LogicalPlan` Computed and cached in stats. Used in stats and verboseStringWithSuffix. Reset in invalidateStatsCache

Getting Cached or Calculating Estimated Statistics — `stats` Method



stats(conf: CatalystConf): Statistics

1

2

3

4

5

stats(conf: CatalystConf): Statistics

stats returns the cached plan statistics or computes a new one (and caches it as statsCache).

Note

stats is used when:

A LogicalPlan computes Statistics
QueryExecution builds complete text representation
JoinSelection checks whether a plan can be broadcast et al
CostBasedJoinReorder attempts to reorder inner joins
LimitPushDown is executed (for FullOuter join)
AggregateEstimation estimates Statistics
FilterEstimation estimates child Statistics
InnerOuterEstimation estimates Statistics of the left and right sides of a join
LeftSemiAntiEstimation estimates Statistics
ProjectEstimation estimates Statistics

`invalidateStatsCache` method

Caution

FIXME

`verboseStringWithSuffix` method

Caution

FIXME

`setAnalyzed` method

Caution

FIXME

Is Logical Plan Streaming? — `isStreaming` method



isStreaming: Boolean

1

2

3

4

5

isStreaming: Boolean

isStreaming is part of the public API of LogicalPlan and is enabled (i.e. true) when a logical plan is a streaming source.

By default, it walks over subtrees and calls itself, i.e. isStreaming, on every child node to find a streaming source.



val spark: SparkSession = ...

// Regular dataset
scala> val ints = spark.createDataset(0 to 9)
ints: org.apache.spark.sql.Dataset[Int] = [value: int]

scala> ints.queryExecution.logical.isStreaming
res1: Boolean = false

// Streaming dataset
scala> val logs = spark.readStream.format("text").load("logs/*.out")
logs: org.apache.spark.sql.DataFrame = [value: string]

scala> logs.queryExecution.logical.isStreaming
res2: Boolean = true

1

2

3

4

5

6

7

8

9

10

11

12

13

14

15

16

17

18

19

val spark: SparkSession = ...

// Regular dataset

scala> val ints = spark.createDataset(0 to 9)

ints: org.apache.spark.sql.Dataset[Int] = [value: int]

scala> ints.queryExecution.logical.isStreaming

res1: Boolean = false

// Streaming dataset

scala> val logs = spark.readStream.format("text").load("logs/*.out")

logs: org.apache.spark.sql.DataFrame = [value: string]

scala> logs.queryExecution.logical.isStreaming

res2: Boolean = true

Note	Streaming Datasets are part of Structured Streaming.

Refreshing Child Logical Plans — `refresh` Method



refresh(): Unit

1

2

3

4

5

refresh(): Unit

refresh calls itself recursively for every child logical operator.

Note	`refresh` is overriden by LogicalRelation only (that refreshes the location of `HadoopFsRelation` relations only).

Note	`refresh` is used when: `SessionCatalog` is requested to refresh a table `CatalogImpl` is requested to refresh a table

`resolveQuoted` Method



resolveQuoted(
  name: String,
  resolver: Resolver): Option[NamedExpression]

1

2

3

4

5

6

7

resolveQuoted(

name: String,

resolver: Resolver): Option[NamedExpression]

resolveQuoted…FIXME

Note	`resolveQuoted` is used when…FIXME

Resolving Attribute By Name Parts — `resolve` Method



resolve(schema: StructType, resolver: Resolver): Seq[Attribute]
resolve(
  nameParts: Seq[String],
  resolver: Resolver): Option[NamedExpression]
resolve(
  nameParts: Seq[String],
  input: Seq[Attribute],
  resolver: Resolver): Option[NamedExpression]  (1)

1

2

3

4

5

6

7

8

9

10

11

12

resolve(schema: StructType, resolver: Resolver): Seq[Attribute]

resolve(

nameParts: Seq[String],

resolver: Resolver): Option[NamedExpression]

resolve(

nameParts: Seq[String],

input: Seq[Attribute],

resolver: Resolver): Option[NamedExpression] (1)

A protected method

resolve…FIXME

Note	`resolve` is used when…FIXME

WindowSpecDefinition

2012-10-12admin阅读(1475)

WindowSpecDefinition Unevaluable Expression

WindowSpecDefinition is an unevaluable expression (i.e. with no support for eval and doGenCode methods).

WindowSpecDefinition is created when:

AstBuilder is requested to parse a window specification in a SQL query
Column.over operator is used



import org.apache.spark.sql.expressions.Window
val byValueDesc = Window.partitionBy("value").orderBy($"value".desc)

val q = table.withColumn("count over window", count("*") over byValueDesc)

import org.apache.spark.sql.catalyst.expressions.WindowExpression
val windowExpr = q.queryExecution
  .logical
  .expressions(1)
  .children(0)
  .asInstanceOf[WindowExpression]

scala> windowExpr.windowSpec
res0: org.apache.spark.sql.catalyst.expressions.WindowSpecDefinition = windowspecdefinition('value, 'value DESC NULLS LAST, UnspecifiedFrame)

1

2

3

4

5

6

7

8

9

10

11

12

13

14

15

16

17

18

import org.apache.spark.sql.expressions.Window

val byValueDesc = Window.partitionBy("value").orderBy($"value".desc)

val q = table.withColumn("count over window", count("*") over byValueDesc)

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

val windowExpr = q.queryExecution

.logical

.expressions(1)

.children(0)

.asInstanceOf[WindowExpression]

scala> windowExpr.windowSpec

res0: org.apache.spark.sql.catalyst.expressions.WindowSpecDefinition = windowspecdefinition('value, 'value DESC NULLS LAST, UnspecifiedFrame)



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

Seq((0, "hello"), (1, "windows"))
  .toDF("id", "token")
  .createOrReplaceTempView("mytable")

val sqlText = """
  SELECT count(*) OVER myWindowSpec
  FROM mytable
  WINDOW
    myWindowSpec AS (
      PARTITION BY token
      ORDER BY id
      RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW
    )
"""

import spark.sessionState.{analyzer,sqlParser}

scala> val parsedPlan = sqlParser.parsePlan(sqlText)
parsedPlan: org.apache.spark.sql.catalyst.plans.logical.LogicalPlan =
'WithWindowDefinition Map(myWindowSpec -> windowspecdefinition('token, 'id ASC NULLS FIRST, RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW))
+- 'Project [unresolvedalias(unresolvedwindowexpression('count(1), WindowSpecReference(myWindowSpec)), None)]
   +- 'UnresolvedRelation `mytable`

import org.apache.spark.sql.catalyst.plans.logical.WithWindowDefinition
val myWindowSpec = parsedPlan.asInstanceOf[WithWindowDefinition].windowDefinitions("myWindowSpec")

scala> println(myWindowSpec)
windowspecdefinition('token, 'id ASC NULLS FIRST, RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW)

scala> println(myWindowSpec.sql)
(PARTITION BY `token` ORDER BY `id` ASC NULLS FIRST RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW)

scala> sql(sqlText)
res4: org.apache.spark.sql.DataFrame = [count(1) OVER (PARTITION BY token ORDER BY id ASC NULLS FIRST RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW): bigint]

scala> println(analyzer.execute(sqlParser.parsePlan(sqlText)))
Project [count(1) OVER (PARTITION BY token ORDER BY id ASC NULLS FIRST RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW)#25L]
+- Project [token#13, id#12, count(1) OVER (PARTITION BY token ORDER BY id ASC NULLS FIRST RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW)#25L, count(1) OVER (PARTITION BY token ORDER BY id ASC NULLS FIRST RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW)#25L]
   +- Window [count(1) windowspecdefinition(token#13, id#12 ASC NULLS FIRST, RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW) AS count(1) OVER (PARTITION BY token ORDER BY id ASC NULLS FIRST RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW)#25L], [token#13], [id#12 ASC NULLS FIRST]
      +- Project [token#13, id#12]
         +- SubqueryAlias mytable
            +- Project [_1#9 AS id#12, _2#10 AS token#13]
               +- LocalRelation [_1#9, _2#10]

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

37

38

39

40

41

42

43

44

45

46

47

48

49

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

Seq((0, "hello"), (1, "windows"))

.toDF("id", "token")

.createOrReplaceTempView("mytable")

val sqlText = """

SELECT count(*) OVER myWindowSpec

FROM mytable

WINDOW

myWindowSpec AS (

PARTITION BY token

ORDER BY id

RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW

)

"""

import spark.sessionState.{analyzer,sqlParser}

scala> val parsedPlan = sqlParser.parsePlan(sqlText)

parsedPlan: org.apache.spark.sql.catalyst.plans.logical.LogicalPlan =

'WithWindowDefinition Map(myWindowSpec -> windowspecdefinition('token, 'id ASC NULLS FIRST, RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW))

+- 'Project [unresolvedalias(unresolvedwindowexpression('count(1), WindowSpecReference(myWindowSpec)), None)]

+- 'UnresolvedRelation `mytable`

import org.apache.spark.sql.catalyst.plans.logical.WithWindowDefinition

val myWindowSpec = parsedPlan.asInstanceOf[WithWindowDefinition].windowDefinitions("myWindowSpec")

scala> println(myWindowSpec)

windowspecdefinition('token, 'id ASC NULLS FIRST, RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW)

scala> println(myWindowSpec.sql)

(PARTITION BY `token` ORDER BY `id` ASC NULLS FIRST RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW)

scala> sql(sqlText)

res4: org.apache.spark.sql.DataFrame = [count(1) OVER (PARTITION BY token ORDER BY id ASC NULLS FIRST RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW): bigint]

scala> println(analyzer.execute(sqlParser.parsePlan(sqlText)))

Project [count(1) OVER (PARTITION BY token ORDER BY id ASC NULLS FIRST RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW)#25L]

+- Project [token#13, id#12, count(1) OVER (PARTITION BY token ORDER BY id ASC NULLS FIRST RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW)#25L, count(1) OVER (PARTITION BY token ORDER BY id ASC NULLS FIRST RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW)#25L]

+- Window [count(1) windowspecdefinition(token#13, id#12 ASC NULLS FIRST, RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW) AS count(1) OVER (PARTITION BY token ORDER BY id ASC NULLS FIRST RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW)#25L], [token#13], [id#12 ASC NULLS FIRST]

+- Project [token#13, id#12]

+- SubqueryAlias mytable

+- Project [_1#9 AS id#12, _2#10 AS token#13]

+- LocalRelation [_1#9, _2#10]

Name Description

children

Window partition and order specifications (for which WindowExpression was created).

dataType

Unsupported (i.e. reports a UnsupportedOperationException)

foldable

Disabled (i.e. false)

nullable

Enabled (i.e. true)

resolved

Enabled when children are and the input DataType is valid and the input frameSpecification is a SpecifiedWindowFrame.

sql

Contains PARTITION BY with comma-separated elements of partitionSpec (if defined) with ORDER BY with comma-separated elements of orderSpec (if defined) followed by frameSpecification.



(PARTITION BY `token` ORDER BY `id` ASC NULLS FIRST RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW)

1

2

3

4

5

(PARTITION BY `token` ORDER BY `id` ASC NULLS FIRST RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW)

Creating WindowSpecDefinition Instance

WindowSpecDefinition takes the following when created:

Expressions for window partition specification
Window order specifications (as SortOrder unary expressions)
Window frame specification (as WindowFrame)

Validating Data Type Of Window Order– `isValidFrameType` Internal Method



isValidFrameType(ft: DataType): Boolean

1

2

3

4

5

isValidFrameType(ft: DataType): Boolean

isValidFrameType is positive (true) when the data type of the window order specification and the input ft data type are as follows:

Otherwise, isValidFrameType is negative (false).

Note	`isValidFrameType` is used exclusively when `WindowSpecDefinition` is requested to checkInputDataTypes (with `RangeFrame` as the window frame specification)

Checking Input Data Types — `checkInputDataTypes` Method



checkInputDataTypes(): TypeCheckResult

1

2

3

4

5

checkInputDataTypes(): TypeCheckResult

Note	`checkInputDataTypes` is part of the Expression Contract to checks the input data types.

checkInputDataTypes…FIXME

WindowFunction Contract — Window Function Expressions With WindowFrame

2012-10-11admin阅读(1385)

WindowFunction Contract — Window Function Expressions With WindowFrame

WindowFunction is the contract of function expressions that define a WindowFrame in which the window operator must be executed.



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

trait WindowFunction extends Expression {
  // No required properties (vals and methods) that have no implementation
}

1

2

3

4

5

6

7

8

9

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

trait WindowFunction extends Expression {

// No required properties (vals and methods) that have no implementation

}

Table 1. WindowFunctions (Direct Implementations)
WindowFunction	Description
AggregateWindowFunction
OffsetWindowFunction

Defining WindowFrame for Execution — `frame` Method



frame: WindowFrame

1

2

3

4

5

frame: WindowFrame

frame defines the WindowFrame for function execution, i.e. the WindowFrame in which the window operator must be executed.

frame is UnspecifiedFrame by default.

Note	`frame` is used when…FIXME

WindowExpression

2012-10-10admin阅读(1459)

WindowExpression Unevaluable Expression

WindowExpression is an unevaluable expression that represents a window function (over some WindowSpecDefinition).

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.

WindowExpression is created when:

WindowSpec is requested to withAggregate (when Column.over operator is used)
WindowsSubstitution logical evaluation rule is executed (with WithWindowDefinition logical operators with UnresolvedWindowExpression expressions)
AstBuilder is requested to parse a function call in a SQL statement

Note	`WindowExpression` can only be created with AggregateExpression, AggregateWindowFunction or OffsetWindowFunction expressions which is enforced at analysis.



// Using Catalyst DSL
val wf = 'count.function(star())
val windowSpec = ???

1

2

3

4

5

6

7

// Using Catalyst DSL

val wf = 'count.function(star())

val windowSpec = ???

Note	`WindowExpression` is resolved in ExtractWindowExpressions, ResolveWindowFrame and ResolveWindowOrder logical rules.



import org.apache.spark.sql.catalyst.expressions.WindowExpression
// relation - Dataset as a table to query
val table = spark.emptyDataset[Int]

scala> val windowExpr = table
  .selectExpr("count() OVER (PARTITION BY value) AS count")
  .queryExecution
  .logical      (1)
  .expressions
  .toList(0)
  .children(0)
  .asInstanceOf[WindowExpression]
windowExpr: org.apache.spark.sql.catalyst.expressions.WindowExpression = 'count() windowspecdefinition('value, UnspecifiedFrame)

scala> windowExpr.sql
res2: String = count() OVER (PARTITION BY `value` UnspecifiedFrame)

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.expressions.WindowExpression

// relation - Dataset as a table to query

val table = spark.emptyDataset[Int]

scala> val windowExpr = table

.selectExpr("count() OVER (PARTITION BY value) AS count")

.queryExecution

.logical (1)

.expressions

.toList(0)

.children(0)

.asInstanceOf[WindowExpression]

windowExpr: org.apache.spark.sql.catalyst.expressions.WindowExpression = 'count() windowspecdefinition('value, UnspecifiedFrame)

scala> windowExpr.sql

res2: String = count() OVER (PARTITION BY `value` UnspecifiedFrame)

Use sqlParser directly as in WithWindowDefinition Example

Table 1. WindowExpression’s Properties
Name	Description
`children`	Collection of two expressions, i.e. windowFunction and WindowSpecDefinition, for which `WindowExpression` was created.
`dataType`	DataType of windowFunction
`foldable`	Whether or not windowFunction is foldable.
`nullable`	Whether or not windowFunction is nullable.
`sql`	`"[windowFunction].sql OVER [windowSpec].sql"`
`toString`	`"[windowFunction] [windowSpec]"`

Note	`WindowExpression` is subject to NullPropagation and DecimalAggregates logical optimizations.

Note	Distinct window functions are not supported which is enforced at analysis.

Note	An offset window function can only be evaluated in an ordered row-based window frame with a single offset which is enforced at analysis.

Catalyst DSL — `windowExpr` Operator



windowExpr(windowFunc: Expression, windowSpec: WindowSpecDefinition): WindowExpression

1

2

3

4

5

windowExpr(windowFunc: Expression, windowSpec: WindowSpecDefinition): WindowExpression

windowExpr operator in Catalyst DSL creates a WindowExpression expression, e.g. for testing or Spark SQL internals exploration.



// FIXME: DEMO

1

2

3

4

5

// FIXME: DEMO

Creating WindowExpression Instance

WindowExpression takes the following when created:

Window function expression
WindowSpecDefinition expression

UnresolvedWindowExpression

2012-10-09admin阅读(2380)

UnresolvedWindowExpression Unevaluable Expression — WindowExpression With Unresolved Window Specification Reference

UnresolvedWindowExpression is an unevaluable expression that represents…FIXME

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.

UnresolvedWindowExpression is created when:

FIXME

UnresolvedWindowExpression is created to represent a child expression and WindowSpecReference (with an identifier for the window reference) when AstBuilder parses a function evaluated in a windowed context with a WindowSpecReference.

UnresolvedWindowExpression is resolved to a WindowExpression when Analyzer resolves UnresolvedWindowExpressions.



import spark.sessionState.sqlParser

scala> sqlParser.parseExpression("foo() OVER windowSpecRef")
res1: org.apache.spark.sql.catalyst.expressions.Expression = unresolvedwindowexpression('foo(), WindowSpecReference(windowSpecRef))

1

2

3

4

5

6

7

8

import spark.sessionState.sqlParser

scala> sqlParser.parseExpression("foo() OVER windowSpecRef")

res1: org.apache.spark.sql.catalyst.expressions.Expression = unresolvedwindowexpression('foo(), WindowSpecReference(windowSpecRef))

Table 1. UnresolvedWindowExpression’s Properties
Name	Description
`dataType`	Reports a `UnresolvedException`
`foldable`	Reports a `UnresolvedException`
`nullable`	Reports a `UnresolvedException`
`resolved`	Disabled (i.e. `false`)

UnresolvedStar

2012-10-08admin阅读(1573)

UnresolvedStar Expression

UnresolvedStar is a Star expression that represents a star (i.e. all) expression in a logical query plan.

UnresolvedStar is created when:

Column is created with *
AstBuilder is requested to visitStar



val q = spark.range(5).select("*")
val plan = q.queryExecution.logical
scala> println(plan.numberedTreeString)
00 'Project [*]
01 +- AnalysisBarrier
02       +- Range (0, 5, step=1, splits=Some(8))

import org.apache.spark.sql.catalyst.analysis.UnresolvedStar
val starExpr = plan.expressions.head.asInstanceOf[UnresolvedStar]

val namedExprs = starExpr.expand(input = q.queryExecution.analyzed, spark.sessionState.analyzer.resolver)
scala> println(namedExprs.head.numberedTreeString)
00 id#0: bigint

1

2

3

4

5

6

7

8

9

10

11

12

13

14

15

16

17

val q = spark.range(5).select("*")

val plan = q.queryExecution.logical

scala> println(plan.numberedTreeString)

00 'Project [*]

01 +- AnalysisBarrier

02 +- Range (0, 5, step=1, splits=Some(8))

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

val starExpr = plan.expressions.head.asInstanceOf[UnresolvedStar]

val namedExprs = starExpr.expand(input = q.queryExecution.analyzed, spark.sessionState.analyzer.resolver)

scala> println(namedExprs.head.numberedTreeString)

00 id#0: bigint

UnresolvedStar can never be resolved, and is expanded at analysis (when ResolveReferences logical resolution rule is executed).

Note	`UnresolvedStar` can only be used in Project, Aggregate or ScriptTransformation logical operators.

Given UnresolvedStar can never be resolved it should not come as a surprise that it cannot be evaluated either (i.e. produce a value given an internal row). When requested to evaluate, UnresolvedStar simply reports a UnsupportedOperationException.



Cannot evaluate expression: [this]

1

2

3

4

5

Cannot evaluate expression: [this]

When created, UnresolvedStar takes name parts that, once concatenated, is the target of the star expansion.



import org.apache.spark.sql.catalyst.analysis.UnresolvedStar
scala> val us = UnresolvedStar(None)
us: org.apache.spark.sql.catalyst.analysis.UnresolvedStar = *

scala> val ab = UnresolvedStar(Some("a" :: "b" :: Nil))
ab: org.apache.spark.sql.catalyst.analysis.UnresolvedStar = List(a, b).*

1

2

3

4

5

6

7

8

9

10

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

scala> val us = UnresolvedStar(None)

us: org.apache.spark.sql.catalyst.analysis.UnresolvedStar = *

scala> val ab = UnresolvedStar(Some("a" :: "b" :: Nil))

ab: org.apache.spark.sql.catalyst.analysis.UnresolvedStar = List(a, b).*

Tip

Use star operator from Catalyst DSL’s expressions to create an UnresolvedStar.



import org.apache.spark.sql.catalyst.dsl.expressions._
val s = star()
scala> :type s
org.apache.spark.sql.catalyst.expressions.Expression

import org.apache.spark.sql.catalyst.analysis.UnresolvedStar
assert(s.isInstanceOf[UnresolvedStar])

val s = star("a", "b")
scala> println(s)
WrappedArray(a, b).*

1

2

3

4

5

6

7

8

9

10

11

12

13

14

15

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

val s = star()

scala> :type s

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

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

assert(s.isInstanceOf[UnresolvedStar])

val s = star("a", "b")

scala> println(s)

WrappedArray(a, b).*

You could also use $"" or ' to create an UnresolvedStar, but that requires sbt console (with Spark libraries defined in build.sbt) as the Catalyst DSL expressions implicits interfere with the Spark implicits to create columns.

Note

AstBuilder replaces count(*) (with no DISTINCT keyword) to count(1).



val q = sql("SELECT COUNT(*) FROM RANGE(1,2,3)")
scala> println(q.queryExecution.logical.numberedTreeString)
00 'Project [unresolvedalias('count(1), None)]
01 +- 'UnresolvedTableValuedFunction range, [1, 2, 3]

val q = sql("SELECT COUNT(DISTINCT *) FROM RANGE(1,2,3)")
scala> println(q.queryExecution.logical.numberedTreeString)
00 'Project [unresolvedalias('COUNT(*), None)]
01 +- 'UnresolvedTableValuedFunction RANGE, [1, 2, 3]

1

2

3

4

5

6

7

8

9

10

11

12

13

val q = sql("SELECT COUNT(*) FROM RANGE(1,2,3)")

scala> println(q.queryExecution.logical.numberedTreeString)

00 'Project [unresolvedalias('count(1), None)]

01 +- 'UnresolvedTableValuedFunction range, [1, 2, 3]

val q = sql("SELECT COUNT(DISTINCT *) FROM RANGE(1,2,3)")

scala> println(q.queryExecution.logical.numberedTreeString)

00 'Project [unresolvedalias('COUNT(*), None)]

01 +- 'UnresolvedTableValuedFunction RANGE, [1, 2, 3]

Star Expansion — `expand` Method



expand(input: LogicalPlan, resolver: Resolver): Seq[NamedExpression]

1

2

3

4

5

expand(input: LogicalPlan, resolver: Resolver): Seq[NamedExpression]

Note	`expand` is part of Star Contract to…FIXME.

expand first expands to named expressions per target:

For unspecified target, expand gives the output schema of the input logical query plan (that assumes that the star refers to a relation / table)
For target with one element, expand gives the table (attribute) in the output schema of the input logical query plan (using qualifiers) if available

With no result earlier, expand then requests the input logical query plan to resolve the target name parts to a named expression.

For a named expression of StructType data type, expand creates an Alias expression with a GetStructField unary expression (with the resolved named expression and the field index).



val q = Seq((0, "zero")).toDF("id", "name").select(struct("id", "name") as "s")
val analyzedPlan = q.queryExecution.analyzed

import org.apache.spark.sql.catalyst.analysis.UnresolvedStar
import org.apache.spark.sql.catalyst.dsl.expressions._
val s = star("s").asInstanceOf[UnresolvedStar]
val exprs = s.expand(input = analyzedPlan, spark.sessionState.analyzer.resolver)

// star("s") should expand to two Alias(GetStructField) expressions
// s is a struct of id and name in the query

import org.apache.spark.sql.catalyst.expressions.{Alias, GetStructField}
val getStructFields = exprs.collect { case Alias(g: GetStructField, _) => g }.map(_.sql)
scala> getStructFields.foreach(println)
`s`.`id`
`s`.`name`

1

2

3

4

5

6

7

8

9

10

11

12

13

14

15

16

17

18

19

20

val q = Seq((0, "zero")).toDF("id", "name").select(struct("id", "name") as "s")

val analyzedPlan = q.queryExecution.analyzed

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

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

val s = star("s").asInstanceOf[UnresolvedStar]

val exprs = s.expand(input = analyzedPlan, spark.sessionState.analyzer.resolver)

// star("s") should expand to two Alias(GetStructField) expressions

// s is a struct of id and name in the query

import org.apache.spark.sql.catalyst.expressions.{Alias, GetStructField}

val getStructFields = exprs.collect { case Alias(g: GetStructField, _) => g }.map(_.sql)

scala> getStructFields.foreach(println)

`s`.`id`

`s`.`name`

expand reports a AnalysisException when:

The data type of the named expression (when the input logical plan was requested to resolve the target) is not a StructType.

Can only star expand struct data types. Attribute: `[target]`

1
2
3
4
5

Can only star expand struct data types. Attribute: `[target]`
Earlier attempts gave no results

cannot resolve '[target].*' given input columns '[from]'

1
2
3
4
5

cannot resolve '[target].*' given input columns '[from]'

UnresolvedRegex

2012-10-07admin阅读(1233)

UnresolvedRegex

UnresolvedRegex is…FIXME

UnresolvedOrdinal

2012-10-06admin阅读(1615)

UnresolvedOrdinal Unevaluable Leaf Expression

UnresolvedOrdinal is a leaf expression that represents a single integer literal in Sort logical operators (in SortOrder ordering expressions) and in Aggregate logical operators (in grouping expressions) in a logical plan.

UnresolvedOrdinal is created when SubstituteUnresolvedOrdinals logical resolution rule is executed.



// Note "order by 1" clause
val sqlText = "select id from VALUES 1, 2, 3 t1(id) order by 1"
val logicalPlan = spark.sql(sqlText).queryExecution.logical
scala> println(logicalPlan.numberedTreeString)
00 'Sort [1 ASC NULLS FIRST], true
01 +- 'Project ['id]
02    +- 'SubqueryAlias t1
03       +- 'UnresolvedInlineTable [id], [List(1), List(2), List(3)]

import org.apache.spark.sql.catalyst.analysis.SubstituteUnresolvedOrdinals
val rule = new SubstituteUnresolvedOrdinals(spark.sessionState.conf)

val logicalPlanWithUnresolvedOrdinals = rule.apply(logicalPlan)
scala> println(logicalPlanWithUnresolvedOrdinals.numberedTreeString)
00 'Sort [unresolvedordinal(1) ASC NULLS FIRST], true
01 +- 'Project ['id]
02    +- 'SubqueryAlias t1
03       +- 'UnresolvedInlineTable [id], [List(1), List(2), List(3)]

import org.apache.spark.sql.catalyst.plans.logical.Sort
val sortOp = logicalPlanWithUnresolvedOrdinals.collect { case s: Sort => s }.head
val sortOrder = sortOp.order.head

import org.apache.spark.sql.catalyst.analysis.UnresolvedOrdinal
val unresolvedOrdinalExpr = sortOrder.child.asInstanceOf[UnresolvedOrdinal]
scala> println(unresolvedOrdinalExpr)
unresolvedordinal(1)

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

// Note "order by 1" clause

val sqlText = "select id from VALUES 1, 2, 3 t1(id) order by 1"

val logicalPlan = spark.sql(sqlText).queryExecution.logical

scala> println(logicalPlan.numberedTreeString)

00 'Sort [1 ASC NULLS FIRST], true

01 +- 'Project ['id]

02 +- 'SubqueryAlias t1

03 +- 'UnresolvedInlineTable [id], [List(1), List(2), List(3)]

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

val rule = new SubstituteUnresolvedOrdinals(spark.sessionState.conf)

val logicalPlanWithUnresolvedOrdinals = rule.apply(logicalPlan)

scala> println(logicalPlanWithUnresolvedOrdinals.numberedTreeString)

00 'Sort [unresolvedordinal(1) ASC NULLS FIRST], true

01 +- 'Project ['id]

02 +- 'SubqueryAlias t1

03 +- 'UnresolvedInlineTable [id], [List(1), List(2), List(3)]

import org.apache.spark.sql.catalyst.plans.logical.Sort

val sortOp = logicalPlanWithUnresolvedOrdinals.collect { case s: Sort => s }.head

val sortOrder = sortOp.order.head

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

val unresolvedOrdinalExpr = sortOrder.child.asInstanceOf[UnresolvedOrdinal]

scala> println(unresolvedOrdinalExpr)

unresolvedordinal(1)

UnresolvedOrdinal takes a single ordinal integer when created.

UnresolvedOrdinal is an unevaluable expression and cannot be evaluated (i.e. produce a value given an internal row).

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.

UnresolvedOrdinal can never be resolved (and is replaced at analysis phase).

Note	`UnresolvedOrdinal` is resolved when ResolveOrdinalInOrderByAndGroupBy logical resolution rule is executed.

UnresolvedOrdinal has no representation in SQL.

Note	`UnresolvedOrdinal` in GROUP BY ordinal position is not allowed for a select list with a star (`*`).

spark-sql 第30页

RunnableCommand Contract — Generic Logical Command with Side Effects

RunnableCommand Contract — Generic Logical Command with Side Effects

Command Contract — Eagerly-Executed Logical Operator

Command Contract — Eagerly-Executed Logical Operator

LogicalPlan Contract — Logical Operator with Children and Expressions / Logical Query Plan

LogicalPlan Contract — Logical Relational Operator with Children and Expressions / Logical Query Plan

Getting Cached or Calculating Estimated Statistics — `stats` Method

`invalidateStatsCache` method

`verboseStringWithSuffix` method

`setAnalyzed` method

Is Logical Plan Streaming? — `isStreaming` method

Refreshing Child Logical Plans — `refresh` Method

`resolveQuoted` Method

Resolving Attribute By Name Parts — `resolve` Method

WindowSpecDefinition

WindowSpecDefinition Unevaluable Expression

Creating WindowSpecDefinition Instance

Validating Data Type Of Window Order– `isValidFrameType` Internal Method

Checking Input Data Types — `checkInputDataTypes` Method

WindowFunction Contract — Window Function Expressions With WindowFrame

WindowFunction Contract — Window Function Expressions With WindowFrame

Defining WindowFrame for Execution — `frame` Method

WindowExpression

WindowExpression Unevaluable Expression

Catalyst DSL — `windowExpr` Operator

Creating WindowExpression Instance

UnresolvedWindowExpression

UnresolvedWindowExpression Unevaluable Expression — WindowExpression With Unresolved Window Specification Reference

UnresolvedStar

UnresolvedStar Expression

Star Expansion — `expand` Method

UnresolvedRegex

UnresolvedRegex

UnresolvedOrdinal

UnresolvedOrdinal Unevaluable Leaf Expression

欢迎关注：spark技术分享

关注公众号：spark技术分享

QQ咨询

回顶部

spark-sql 第30页

RunnableCommand Contract — Generic Logical Command with Side Effects

Command Contract — Eagerly-Executed Logical Operator

LogicalPlan Contract — Logical Relational Operator with Children and Expressions / Logical Query Plan

Getting Cached or Calculating Estimated Statistics — stats Method

invalidateStatsCache method

verboseStringWithSuffix method

setAnalyzed method

Is Logical Plan Streaming? — isStreaming method

Refreshing Child Logical Plans — refresh Method

resolveQuoted Method

Resolving Attribute By Name Parts — resolve Method

WindowSpecDefinition Unevaluable Expression

Creating WindowSpecDefinition Instance

Validating Data Type Of Window Order– isValidFrameType Internal Method

Checking Input Data Types — checkInputDataTypes Method

WindowFunction Contract — Window Function Expressions With WindowFrame

Defining WindowFrame for Execution — frame Method

WindowExpression Unevaluable Expression

Catalyst DSL — windowExpr Operator

Creating WindowExpression Instance

UnresolvedWindowExpression Unevaluable Expression — WindowExpression With Unresolved Window Specification Reference

UnresolvedStar Expression

Star Expansion — expand Method

UnresolvedRegex

UnresolvedOrdinal Unevaluable Leaf Expression

欢迎关注：spark技术分享

关注公众号：spark技术分享

QQ咨询

回顶部

Getting Cached or Calculating Estimated Statistics — `stats` Method

`invalidateStatsCache` method

`verboseStringWithSuffix` method

`setAnalyzed` method

Is Logical Plan Streaming? — `isStreaming` method

Refreshing Child Logical Plans — `refresh` Method

`resolveQuoted` Method

Resolving Attribute By Name Parts — `resolve` Method

Validating Data Type Of Window Order– `isValidFrameType` Internal Method

Checking Input Data Types — `checkInputDataTypes` Method

Defining WindowFrame for Execution — `frame` Method

Catalyst DSL — `windowExpr` Operator

Star Expansion — `expand` Method