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

UnresolvedInlineTable Logical Operator

UnresolvedInlineTable is a unary logical operator that represents an inline table (aka virtual table in Apache Hive).

UnresolvedInlineTable is created when AstBuilder is requested to parse an inline table in a SQL statement.



// `tableAlias` undefined so columns default to `colN`
val q = sql("VALUES 1, 2, 3")
scala> println(q.queryExecution.logical.numberedTreeString)
00 'UnresolvedInlineTable [col1], [List(1), List(2), List(3)]

// CreateNamedStruct with `tableAlias`
val q = sql("VALUES (1, 'a'), (2, 'b') AS t1(a, b)")
scala> println(q.queryExecution.logical.numberedTreeString)
00 'SubqueryAlias t1
01 +- 'UnresolvedInlineTable [a, b], [List(1, a), List(2, b)]

1

2

3

4

5

6

7

8

9

10

11

12

13

14

// `tableAlias` undefined so columns default to `colN`

val q = sql("VALUES 1, 2, 3")

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

00 'UnresolvedInlineTable [col1], [List(1), List(2), List(3)]

// CreateNamedStruct with `tableAlias`

val q = sql("VALUES (1, 'a'), (2, 'b') AS t1(a, b)")

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

00 'SubqueryAlias t1

01 +- 'UnresolvedInlineTable [a, b], [List(1, a), List(2, b)]

UnresolvedInlineTable is never resolved (and is converted to a LocalRelation in ResolveInlineTables logical resolution rule).

UnresolvedInlineTable uses no output schema attributes.

UnresolvedInlineTable uses expressionsResolved flag that is on (true) only when all the Catalyst expressions in the rows are resolved.

Creating UnresolvedInlineTable Instance

UnresolvedInlineTable takes the following when created:

Column names
Rows (as Catalyst expressions for every row, i.e. Seq[Seq[Expression]])

UnresolvedHint Unary Logical Operator — Attaching Hint to Logical Plan

UnresolvedHint is a unary logical operator that represents a hint (by name and parameters) for the child logical plan.

UnresolvedHint is created and added to a logical plan when:

Dataset.hint operator is used
AstBuilder converts /*+ hint */ in SELECT SQL queries



// Dataset API
val q = spark.range(1).hint("myHint", 100, true)
val plan = q.queryExecution.logical
scala> println(plan.numberedTreeString)
00 'UnresolvedHint myHint, [100, true]
01 +- Range (0, 1, step=1, splits=Some(8))

// SQL
val q = sql("SELECT /*+ myHint (100, true) */ 1")
val plan = q.queryExecution.logical
scala> println(plan.numberedTreeString)
00 'UnresolvedHint myHint, [100, true]
01 +- 'Project [unresolvedalias(1, None)]
02    +- OneRowRelation

1

2

3

4

5

6

7

8

9

10

11

12

13

14

15

16

17

18

// Dataset API

val q = spark.range(1).hint("myHint", 100, true)

val plan = q.queryExecution.logical

scala> println(plan.numberedTreeString)

00 'UnresolvedHint myHint, [100, true]

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

// SQL

val q = sql("SELECT /*+ myHint (100, true) */ 1")

val plan = q.queryExecution.logical

scala> println(plan.numberedTreeString)

00 'UnresolvedHint myHint, [100, true]

01 +- 'Project [unresolvedalias(1, None)]

02 +- OneRowRelation

When created UnresolvedHint takes:

Name of a hint
Parameters of a hint
Child logical plan

UnresolvedHint can never be resolved and is supposed to be converted to a ResolvedHint unary logical operator during query analysis (or simply removed from a logical plan).

Note

There are the following logical rules that Spark Analyzer uses to analyze logical plans with the UnresolvedHint logical operator:

ResolveBroadcastHints resolves UnresolvedHint operators with BROADCAST, BROADCASTJOIN, MAPJOIN hints to a ResolvedHint
ResolveCoalesceHints resolves UnresolvedHint logical operators with COALESCE or REPARTITION hints
RemoveAllHints simply removes all UnresolvedHint operators

The order of executing the above rules matters.



// Let's hint the query twice
// The order of hints matters as every hint operator executes Spark analyzer
// That will resolve all but the last hint
val q = spark.range(100).
  hint("broadcast").
  hint("myHint", 100, true)
val plan = q.queryExecution.logical
scala> println(plan.numberedTreeString)
00 'UnresolvedHint myHint, [100, true]
01 +- ResolvedHint (broadcast)
02    +- Range (0, 100, step=1, splits=Some(8))

// Let's resolve unresolved hints
import org.apache.spark.sql.catalyst.rules.RuleExecutor
import org.apache.spark.sql.catalyst.plans.logical.LogicalPlan
import org.apache.spark.sql.catalyst.analysis.ResolveHints
import org.apache.spark.sql.internal.SQLConf
object HintResolver extends RuleExecutor[LogicalPlan] {
  lazy val batches =
    Batch("Hints", FixedPoint(maxIterations = 100),
      new ResolveHints.ResolveBroadcastHints(SQLConf.get),
      ResolveHints.RemoveAllHints) :: Nil
}
val resolvedPlan = HintResolver.execute(plan)
scala> println(resolvedPlan.numberedTreeString)
00 ResolvedHint (broadcast)
01 +- Range (0, 100, step=1, splits=Some(8))

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

// Let's hint the query twice

// The order of hints matters as every hint operator executes Spark analyzer

// That will resolve all but the last hint

val q = spark.range(100).

hint("broadcast").

hint("myHint", 100, true)

val plan = q.queryExecution.logical

scala> println(plan.numberedTreeString)

00 'UnresolvedHint myHint, [100, true]

01 +- ResolvedHint (broadcast)

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

// Let's resolve unresolved hints

import org.apache.spark.sql.catalyst.rules.RuleExecutor

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

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

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

object HintResolver extends RuleExecutor[LogicalPlan] {

lazy val batches =

Batch("Hints", FixedPoint(maxIterations = 100),

new ResolveHints.ResolveBroadcastHints(SQLConf.get),

ResolveHints.RemoveAllHints) :: Nil

}

val resolvedPlan = HintResolver.execute(plan)

scala> println(resolvedPlan.numberedTreeString)

00 ResolvedHint (broadcast)

01 +- Range (0, 100, step=1, splits=Some(8))

UnresolvedHint uses the child operator’s output schema for yours.

Tip

Use hint operator from Catalyst DSL to create a UnresolvedHint logical operator, e.g. for testing or Spark SQL internals exploration.



// Create a logical plan to add hint to
import org.apache.spark.sql.catalyst.plans.logical.LocalRelation
val r1 = LocalRelation('a.int, 'b.timestamp, 'c.boolean)
scala> println(r1.numberedTreeString)
00 LocalRelation <empty>, [a#0, b#1, c#2]

// Attach hint to the plan
import org.apache.spark.sql.catalyst.dsl.plans._
val plan = r1.hint(name = "myHint", 100, true)
scala> println(plan.numberedTreeString)
00 'UnresolvedHint myHint, [100, true]
01 +- LocalRelation <empty>, [a#0, b#1, c#2]

1

2

3

4

5

6

7

8

9

10

11

12

13

14

15

16

// Create a logical plan to add hint to

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

val r1 = LocalRelation('a.int, 'b.timestamp, 'c.boolean)

scala> println(r1.numberedTreeString)

00 LocalRelation <empty>, [a#0, b#1, c#2]

// Attach hint to the plan

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

val plan = r1.hint(name = "myHint", 100, true)

scala> println(plan.numberedTreeString)

00 'UnresolvedHint myHint, [100, true]

01 +- LocalRelation <empty>, [a#0, b#1, c#2]

UnresolvedCatalogRelation Leaf Logical Operator — Placeholder of Catalog Tables

UnresolvedCatalogRelation is a leaf logical operator that acts as a placeholder in a logical query plan until FindDataSourceTable logical evaluation rule resolves it to a concrete relation logical plan (i.e. a LogicalRelation for a data source table or a HiveTableRelation for hive table).

UnresolvedCatalogRelation is created when SessionCatalog is requested to find a relation (for DescribeTableCommand logical command or ResolveRelations logical evaluation rule).



// non-temporary global or local view
// database defined
// externalCatalog.getTable returns non-VIEW table
// Make the example reproducible
val tableName = "t1"
spark.sharedState.externalCatalog.dropTable(
  db = "default",
  table = tableName,
  ignoreIfNotExists = true,
  purge = true)
spark.range(10).write.saveAsTable(tableName)

scala> :type spark.sessionState.catalog
org.apache.spark.sql.catalyst.catalog.SessionCatalog

import org.apache.spark.sql.catalyst.TableIdentifier
val plan = spark.sessionState.catalog.lookupRelation(TableIdentifier(tableName))
scala> println(plan.numberedTreeString)
00 'SubqueryAlias t1
01 +- 'UnresolvedCatalogRelation `default`.`t1`, org.apache.hadoop.hive.ql.io.parquet.serde.ParquetHiveSerDe

1

2

3

4

5

6

7

8

9

10

11

12

13

14

15

16

17

18

19

20

21

22

23

24

// non-temporary global or local view

// database defined

// externalCatalog.getTable returns non-VIEW table

// Make the example reproducible

val tableName = "t1"

spark.sharedState.externalCatalog.dropTable(

db = "default",

table = tableName,

ignoreIfNotExists = true,

purge = true)

spark.range(10).write.saveAsTable(tableName)

scala> :type spark.sessionState.catalog

org.apache.spark.sql.catalyst.catalog.SessionCatalog

import org.apache.spark.sql.catalyst.TableIdentifier

val plan = spark.sessionState.catalog.lookupRelation(TableIdentifier(tableName))

scala> println(plan.numberedTreeString)

00 'SubqueryAlias t1

01 +- 'UnresolvedCatalogRelation `default`.`t1`, org.apache.hadoop.hive.ql.io.parquet.serde.ParquetHiveSerDe

When created, UnresolvedCatalogRelation asserts that the database is specified.

UnresolvedCatalogRelation can never be resolved and is converted to a LogicalRelation for a data source table or a HiveTableRelation for hive table at analysis phase.

UnresolvedCatalogRelation uses an empty output schema.

UnresolvedCatalogRelation takes a single CatalogTable when created.

Union

2012-12-11admin阅读(1040)

Union Logical Operator

Union is…FIXME

TypedFilter

2012-12-10admin阅读(1302)

TypedFilter Logical Operator

TypedFilter is…FIXME

SubqueryAlias

2012-12-09admin阅读(1934)

SubqueryAlias Unary Logical Operator

SubqueryAlias is a unary logical operator that represents an aliased subquery (i.e. the child logical query plan with the alias in the output schema).

SubqueryAlias is created when:

AstBuilder is requested to parse a named or aliased query, aliased query plan and mayApplyAliasPlan in a SQL statement
Dataset.as operator is used
SessionCatalog is requested to find a table or view in catalogs
RewriteCorrelatedScalarSubquery logical optimization is requested to constructLeftJoins (when applied to Aggregate, Project or Filter logical operators with correlated scalar subqueries)

SubqueryAlias simply requests the child logical operator for the canonicalized version.

When requested for output schema attributes, SubqueryAlias requests the child logical operator for them and adds the alias as a qualifier.

Note	EliminateSubqueryAliases logical optimization eliminates (removes) `SubqueryAlias` operators from a logical query plan.

Note	RewriteCorrelatedScalarSubquery logical optimization rewrites correlated scalar subqueries with `SubqueryAlias` operators.

Catalyst DSL — `subquery` And `as` Operators



as(alias: String): LogicalPlan
subquery(alias: Symbol): LogicalPlan

1

2

3

4

5

6

as(alias: String): LogicalPlan

subquery(alias: Symbol): LogicalPlan

subquery and as operators in Catalyst DSL create a SubqueryAlias logical operator, e.g. for testing or Spark SQL internals exploration.



import org.apache.spark.sql.catalyst.dsl.plans._
val t1 = table("t1")

val plan = t1.subquery('a)
scala> println(plan.numberedTreeString)
00 'SubqueryAlias a
01 +- 'UnresolvedRelation `t1`

val plan = t1.as("a")
scala> println(plan.numberedTreeString)
00 'SubqueryAlias a
01 +- 'UnresolvedRelation `t1`

1

2

3

4

5

6

7

8

9

10

11

12

13

14

15

16

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

val t1 = table("t1")

val plan = t1.subquery('a)

scala> println(plan.numberedTreeString)

00 'SubqueryAlias a

01 +- 'UnresolvedRelation `t1`

val plan = t1.as("a")

scala> println(plan.numberedTreeString)

00 'SubqueryAlias a

01 +- 'UnresolvedRelation `t1`

Creating SubqueryAlias Instance

SubqueryAlias takes the following when created:

Alias
Child logical plan

Sort

2012-12-08admin阅读(1707)

Sort Unary Logical Operator

Sort is a unary logical operator that represents the following in a logical plan:

ORDER BY, SORT BY, SORT BY … DISTRIBUTE BY and CLUSTER BY clauses (when AstBuilder is requested to parse a query)
Dataset.sortWithinPartitions, Dataset.sort and Dataset.randomSplit operators



// Using the feature of ordinal literal
val ids = Seq(1,3,2).toDF("id").sort(lit(1))
val logicalPlan = ids.queryExecution.logical
scala> println(logicalPlan.numberedTreeString)
00 Sort [1 ASC NULLS FIRST], true
01 +- AnalysisBarrier
02       +- Project [value#22 AS id#24]
03          +- LocalRelation [value#22]

import org.apache.spark.sql.catalyst.plans.logical.Sort
val sortOp = logicalPlan.collect { case s: Sort => s }.head
scala> println(sortOp.numberedTreeString)
00 Sort [1 ASC NULLS FIRST], true
01 +- AnalysisBarrier
02       +- Project [value#22 AS id#24]
03          +- LocalRelation [value#22]

1

2

3

4

5

6

7

8

9

10

11

12

13

14

15

16

17

18

19

20

// Using the feature of ordinal literal

val ids = Seq(1,3,2).toDF("id").sort(lit(1))

val logicalPlan = ids.queryExecution.logical

scala> println(logicalPlan.numberedTreeString)

00 Sort [1 ASC NULLS FIRST], true

01 +- AnalysisBarrier

02 +- Project [value#22 AS id#24]

03 +- LocalRelation [value#22]

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

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

scala> println(sortOp.numberedTreeString)

00 Sort [1 ASC NULLS FIRST], true

01 +- AnalysisBarrier

02 +- Project [value#22 AS id#24]

03 +- LocalRelation [value#22]



val nums = Seq((0, "zero"), (1, "one")).toDF("id", "name")
// Creates a Sort logical operator:
// - descending sort direction for id column (specified explicitly)
// - name column is wrapped with ascending sort direction
val numsOrdered = nums.sort('id.desc, 'name)
val logicalPlan = numsOrdered.queryExecution.logical
scala> println(logicalPlan.numberedTreeString)
00 'Sort ['id DESC NULLS LAST, 'name ASC NULLS FIRST], true
01 +- Project [_1#11 AS id#14, _2#12 AS name#15]
02    +- LocalRelation [_1#11, _2#12]

1

2

3

4

5

6

7

8

9

10

11

12

13

14

val nums = Seq((0, "zero"), (1, "one")).toDF("id", "name")

// Creates a Sort logical operator:

// - descending sort direction for id column (specified explicitly)

// - name column is wrapped with ascending sort direction

val numsOrdered = nums.sort('id.desc, 'name)

val logicalPlan = numsOrdered.queryExecution.logical

scala> println(logicalPlan.numberedTreeString)

00 'Sort ['id DESC NULLS LAST, 'name ASC NULLS FIRST], true

01 +- Project [_1#11 AS id#14, _2#12 AS name#15]

02 +- LocalRelation [_1#11, _2#12]

Sort takes the following when created:

SortOrder ordering expressions
global flag for global (true) or partition-only (false) sorting
Child logical plan

The output schema of a Sort operator is the output of the child logical operator.

The maxRows of a Sort operator is the maxRows of the child logical operator.

Tip	Use orderBy or sortBy operators from the Catalyst DSL to create a `Sort` logical operator, e.g. for testing or Spark SQL internals exploration.

Note	Sorting is supported for columns of orderable type only (which is enforced at analysis when `CheckAnalysis` is requested to checkAnalysis).

Note	`Sort` logical operator is resolved to SortExec unary physical operator when BasicOperators execution planning strategy is executed.

Catalyst DSL — `orderBy` and `sortBy` Operators



orderBy(sortExprs: SortOrder*): LogicalPlan
sortBy(sortExprs: SortOrder*): LogicalPlan

1

2

3

4

5

6

orderBy(sortExprs: SortOrder*): LogicalPlan

sortBy(sortExprs: SortOrder*): LogicalPlan

orderBy and sortBy create a Sort logical operator with the global flag on and off, respectively.



import org.apache.spark.sql.catalyst.dsl.plans._
val t1 = table("t1")

import org.apache.spark.sql.catalyst.dsl.expressions._
val globalSortById = t1.orderBy('id.asc_nullsLast)

// Note true for the global flag
scala> println(globalSortById.numberedTreeString)
00 'Sort ['id ASC NULLS LAST], true
01 +- 'UnresolvedRelation `t1`

// Note false for the global flag
val partitionOnlySortById = t1.sortBy('id.asc_nullsLast)
scala> println(partitionOnlySortById.numberedTreeString)
00 'Sort ['id ASC NULLS LAST], false
01 +- 'UnresolvedRelation `t1`

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.plans._

val t1 = table("t1")

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

val globalSortById = t1.orderBy('id.asc_nullsLast)

// Note true for the global flag

scala> println(globalSortById.numberedTreeString)

00 'Sort ['id ASC NULLS LAST], true

01 +- 'UnresolvedRelation `t1`

// Note false for the global flag

val partitionOnlySortById = t1.sortBy('id.asc_nullsLast)

scala> println(partitionOnlySortById.numberedTreeString)

00 'Sort ['id ASC NULLS LAST], false

01 +- 'UnresolvedRelation `t1`

ShowTablesCommand

2012-12-07admin阅读(1496)

ShowTablesCommand Logical Command

ShowTablesCommand is a logical command for…FIXME

Executing Logical Command — `run` Method



run(sparkSession: SparkSession): Seq[Row]

1

2

3

4

5

run(sparkSession: SparkSession): Seq[Row]

Note	`run` is part of RunnableCommand Contract to execute (run) a logical command.

run…FIXME

ShowCreateTableCommand

2012-12-06admin阅读(1452)

ShowCreateTableCommand Logical Command

ShowCreateTableCommand is a logical command that executes a SHOW CREATE TABLE SQL statement (with a data source / non-Hive or a Hive table).

ShowCreateTableCommand is created when SparkSqlAstBuilder is requested to parse SHOW CREATE TABLE SQL statement.

ShowCreateTableCommand uses a single createtab_stmt column (of type StringType) for the output schema.



import org.apache.spark.sql.SaveMode
spark.range(10e4.toLong)
  .write
  .bucketBy(4, "id")
  .sortBy("id")
  .mode(SaveMode.Overwrite)
  .saveAsTable("bucketed_4_10e4")
scala> sql("SHOW CREATE TABLE bucketed_4_10e4").show(truncate = false)
+----------------------------------------------------------------------------------------------------------------------------------------------------+
|createtab_stmt                                                                                                                                      |
+----------------------------------------------------------------------------------------------------------------------------------------------------+
|CREATE TABLE `bucketed_4_10e4` (`id` BIGINT)
USING parquet
OPTIONS (
  `serialization.format` '1'
)
CLUSTERED BY (id)
SORTED BY (id)
INTO 4 BUCKETS
|
+----------------------------------------------------------------------------------------------------------------------------------------------------+

scala> sql("SHOW CREATE TABLE bucketed_4_10e4").as[String].collect.foreach(println)
CREATE TABLE `bucketed_4_10e4` (`id` BIGINT)
USING parquet
OPTIONS (
  `serialization.format` '1'
)
CLUSTERED BY (id)
SORTED BY (id)
INTO 4 BUCKETS

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

import org.apache.spark.sql.SaveMode

spark.range(10e4.toLong)

.write

.bucketBy(4, "id")

.sortBy("id")

.mode(SaveMode.Overwrite)

.saveAsTable("bucketed_4_10e4")

scala> sql("SHOW CREATE TABLE bucketed_4_10e4").show(truncate = false)

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

|createtab_stmt |

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

|CREATE TABLE `bucketed_4_10e4` (`id` BIGINT)

USING parquet

OPTIONS (

`serialization.format` '1'

)

CLUSTERED BY (id)

SORTED BY (id)

INTO 4 BUCKETS

|

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

scala> sql("SHOW CREATE TABLE bucketed_4_10e4").as[String].collect.foreach(println)

CREATE TABLE `bucketed_4_10e4` (`id` BIGINT)

USING parquet

OPTIONS (

`serialization.format` '1'

)

CLUSTERED BY (id)

SORTED BY (id)

INTO 4 BUCKETS

ShowCreateTableCommand takes a single TableIdentifier when created.

Executing Logical Command — `run` Method



run(sparkSession: SparkSession): Seq[Row]

1

2

3

4

5

run(sparkSession: SparkSession): Seq[Row]

Note	`run` is part of RunnableCommand Contract to execute (run) a logical command.

run requests the SparkSession for the SessionState that is used to access the SessionCatalog.

run then requests the SessionCatalog to retrieve the table metadata from the external catalog (metastore).

run then showCreateDataSourceTable for a data source / non-Hive table or showCreateHiveTable for a Hive table (per the table metadata).

In the end, run returns the CREATE TABLE statement in a single Row.

`showHiveTableNonDataColumns` Internal Method



showHiveTableNonDataColumns(metadata: CatalogTable, builder: StringBuilder): Unit

1

2

3

4

5

showHiveTableNonDataColumns(metadata: CatalogTable, builder: StringBuilder): Unit

showHiveTableNonDataColumns…FIXME

Note	`showHiveTableNonDataColumns` is used exclusively when `ShowCreateTableCommand` logical command is requested to showCreateHiveTable.

`showCreateHiveTable` Internal Method



showCreateHiveTable(metadata: CatalogTable): String

1

2

3

4

5

showCreateHiveTable(metadata: CatalogTable): String

showCreateHiveTable…FIXME

Note	`showCreateHiveTable` is used exclusively when `ShowCreateTableCommand` logical command is executed (with a Hive table).

`showHiveTableHeader` Internal Method



showHiveTableHeader(metadata: CatalogTable, builder: StringBuilder): Unit

1

2

3

4

5

showHiveTableHeader(metadata: CatalogTable, builder: StringBuilder): Unit

showHiveTableHeader…FIXME

Note	`showHiveTableHeader` is used exclusively when `ShowCreateTableCommand` logical command is requested to showCreateHiveTable.

SaveIntoDataSourceCommand

2012-12-05admin阅读(2293)

SaveIntoDataSourceCommand Logical Command

SaveIntoDataSourceCommand is a logical command that, when executed, FIXME.

SaveIntoDataSourceCommand is created exclusively when DataSource is requested to create a logical command for writing (to a CreatableRelationProvider data source).

SaveIntoDataSourceCommand returns the logical query plan when requested for the inner nodes (that should be shown as an inner nested tree of this node).



// DEMO Example with inner nodes that should be shown as an inner nested tree of this node

val lines = Seq("SaveIntoDataSourceCommand").toDF("line")

// NOTE: There are two CreatableRelationProviders: jdbc and kafka
// jdbc is simpler to use in spark-shell as it does not need --packages
val url = "jdbc:derby:memory:;databaseName=/tmp/test;create=true"
val requiredOpts = Map("url" -> url, "dbtable" -> "lines")
// Use overwrite SaveMode to make the demo reproducible
import org.apache.spark.sql.SaveMode.Overwrite
lines.write.options(requiredOpts).format("jdbc").mode(Overwrite).save

// Go to web UI's SQL tab and see the last executed query

1

2

3

4

5

6

7

8

9

10

11

12

13

14

15

16

17

// DEMO Example with inner nodes that should be shown as an inner nested tree of this node

val lines = Seq("SaveIntoDataSourceCommand").toDF("line")

// NOTE: There are two CreatableRelationProviders: jdbc and kafka

// jdbc is simpler to use in spark-shell as it does not need --packages

val url = "jdbc:derby:memory:;databaseName=/tmp/test;create=true"

val requiredOpts = Map("url" -> url, "dbtable" -> "lines")

// Use overwrite SaveMode to make the demo reproducible

import org.apache.spark.sql.SaveMode.Overwrite

lines.write.options(requiredOpts).format("jdbc").mode(Overwrite).save

// Go to web UI's SQL tab and see the last executed query

SaveIntoDataSourceCommand redacts the options for the simple description with state prefix.



SaveIntoDataSourceCommand [dataSource], [redacted], [mode]

1

2

3

4

5

SaveIntoDataSourceCommand [dataSource], [redacted], [mode]

Executing Logical Command — `run` Method



run(sparkSession: SparkSession): Seq[Row]

1

2

3

4

5

run(sparkSession: SparkSession): Seq[Row]

Note	`run` is part of RunnableCommand Contract to execute (run) a logical command.

run simply requests the CreatableRelationProvider data source to save the rows of a structured query (a DataFrame).

In the end, run returns an empty Seq[Row] (just to follow the signature and please the Scala compiler).

Creating SaveIntoDataSourceCommand Instance

SaveIntoDataSourceCommand takes the following when created:

Logical query plan
CreatableRelationProvider data source
Options (as Map[String, String])
SaveMode

spark-sql 第24页

UnresolvedInlineTable

UnresolvedInlineTable Logical Operator

Creating UnresolvedInlineTable Instance

UnresolvedHint

UnresolvedHint Unary Logical Operator — Attaching Hint to Logical Plan

UnresolvedCatalogRelation

UnresolvedCatalogRelation Leaf Logical Operator — Placeholder of Catalog Tables

Union

Union Logical Operator

TypedFilter

TypedFilter Logical Operator

SubqueryAlias

SubqueryAlias Unary Logical Operator

Catalyst DSL — `subquery` And `as` Operators

Creating SubqueryAlias Instance

Sort

Sort Unary Logical Operator

Catalyst DSL — `orderBy` and `sortBy` Operators

ShowTablesCommand

ShowTablesCommand Logical Command

Executing Logical Command — `run` Method

ShowCreateTableCommand

ShowCreateTableCommand Logical Command

Executing Logical Command — `run` Method

`showHiveTableNonDataColumns` Internal Method

`showCreateHiveTable` Internal Method

`showHiveTableHeader` Internal Method

SaveIntoDataSourceCommand

SaveIntoDataSourceCommand Logical Command

Executing Logical Command — `run` Method

Creating SaveIntoDataSourceCommand Instance

欢迎关注：spark技术分享

关注公众号：spark技术分享

QQ咨询

回顶部

spark-sql 第24页

UnresolvedInlineTable Logical Operator

Creating UnresolvedInlineTable Instance

UnresolvedHint Unary Logical Operator — Attaching Hint to Logical Plan

UnresolvedCatalogRelation Leaf Logical Operator — Placeholder of Catalog Tables

Union Logical Operator

TypedFilter Logical Operator

SubqueryAlias Unary Logical Operator

Catalyst DSL — subquery And as Operators

Creating SubqueryAlias Instance

Sort Unary Logical Operator

Catalyst DSL — orderBy and sortBy Operators

ShowTablesCommand Logical Command

Executing Logical Command — run Method

ShowCreateTableCommand Logical Command

Executing Logical Command — run Method

showHiveTableNonDataColumns Internal Method

showCreateHiveTable Internal Method

showHiveTableHeader Internal Method

SaveIntoDataSourceCommand Logical Command

Executing Logical Command — run Method

Creating SaveIntoDataSourceCommand Instance

欢迎关注：spark技术分享

关注公众号：spark技术分享

QQ咨询

回顶部

Catalyst DSL — `subquery` And `as` Operators

Catalyst DSL — `orderBy` and `sortBy` Operators

Executing Logical Command — `run` Method

Executing Logical Command — `run` Method

`showHiveTableNonDataColumns` Internal Method

`showCreateHiveTable` Internal Method

`showHiveTableHeader` Internal Method

Executing Logical Command — `run` Method