CreateDataSourceTableAsSelectCommand is a logical command that FIXME.
spark技术分享run Method|
1 2 3 4 5 |
run(session: SparkSession): Seq[Row] |
|
Note
|
run is part of RunnableCommand Contract to execute (run) a logical command.
|
run…FIXME
ClearCacheCommand is a logical command to remove all cached tables from the in-memory cache.
ClearCacheCommand corresponds to CLEAR CACHE SQL statement.
|
Note
|
ClearCacheCommand is described by clearCache labeled alternative in statement expression in SqlBase.g4 and parsed using SparkSqlParser.
|
AnalyzeTableCommand is a logical command that computes statistics (i.e. total size and row count) for a table and stores the stats in a metastore.
AnalyzeTableCommand is created exclusively for ANALYZE TABLE with no PARTITION specification and FOR COLUMNS clause.
|
1 2 3 4 5 6 7 8 9 10 11 |
// Seq((0, 0, "zero"), (1, 1, "one")).toDF("id", "p1", "p2").write.partitionBy("p1", "p2").saveAsTable("t1") val sqlText = "ANALYZE TABLE t1 COMPUTE STATISTICS NOSCAN" val plan = spark.sql(sqlText).queryExecution.logical import org.apache.spark.sql.execution.command.AnalyzeTableCommand val cmd = plan.asInstanceOf[AnalyzeTableCommand] scala> println(cmd) AnalyzeTableCommand `t1`, false |
run Method|
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 session-specific SessionCatalog for the metadata of the table and makes sure that it is not a view (aka temporary table).
|
Note
|
run uses the input SparkSession to access the session-specific SessionState that in turn gives access to the current SessionCatalog.
|
run computes the total size and, without NOSCAN flag, the row count statistics of the table.
|
Note
|
run uses SparkSession to find the table in a metastore.
|
In the end, run alters table statistics if different from the existing table statistics in metastore.
run throws a AnalysisException when executed on a view.
|
1 2 3 4 5 |
ANALYZE TABLE is not supported on views. |
|
Note
|
Row count statistics triggers a Spark job to count the number of rows in a table (that happens with
|
AnalyzeTableCommand takes the following when created:
noscan flag (enabled by default) that indicates whether NOSCAN option was used or not
AnalyzePartitionCommand is a logical command that computes statistics (i.e. total size and row count) for table partitions and stores the stats in a metastore.
AnalyzePartitionCommand is created exclusively for ANALYZE TABLE with PARTITION specification only (i.e. no FOR COLUMNS clause).
|
1 2 3 4 5 6 7 8 9 10 11 |
// Seq((0, 0, "zero"), (1, 1, "one")).toDF("id", "p1", "p2").write.partitionBy("p1", "p2").saveAsTable("t1") val analyzeTable = "ANALYZE TABLE t1 PARTITION (p1, p2) COMPUTE STATISTICS" val plan = spark.sql(analyzeTable).queryExecution.logical import org.apache.spark.sql.execution.command.AnalyzePartitionCommand val cmd = plan.asInstanceOf[AnalyzePartitionCommand] scala> println(cmd) AnalyzePartitionCommand `t1`, Map(p1 -> None, p2 -> None), false |
run Method|
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 session-specific SessionCatalog for the metadata of the table and makes sure that it is not a view.
|
Note
|
run uses the input SparkSession to access the session-specific SessionState that in turn is used to access the current SessionCatalog.
|
run getPartitionSpec.
run requests the session-specific SessionCatalog for the partitions per the partition specification.
run finishes when the table has no partitions defined in a metastore.
run computes row count statistics per partition unless noscan flag was enabled.
run calculates total size (in bytes) (aka partition location size) for every table partition and creates a CatalogStatistics with the current statistics if different from the statistics recorded in the metastore (with a new row count statistic computed earlier).
In the end, run alters table partition metadata for partitions with the statistics changed.
run reports a NoSuchPartitionException when partitions do not match the metastore.
run reports an AnalysisException when executed on a view.
|
1 2 3 4 5 |
ANALYZE TABLE is not supported on views. |
calculateRowCountsPerPartition Internal Method|
1 2 3 4 5 6 7 8 |
calculateRowCountsPerPartition( sparkSession: SparkSession, tableMeta: CatalogTable, partitionValueSpec: Option[TablePartitionSpec]): Map[TablePartitionSpec, BigInt] |
calculateRowCountsPerPartition…FIXME
|
Note
|
calculateRowCountsPerPartition is used exclusively when AnalyzePartitionCommand is executed.
|
getPartitionSpec Internal Method|
1 2 3 4 5 |
getPartitionSpec(table: CatalogTable): Option[TablePartitionSpec] |
getPartitionSpec…FIXME
|
Note
|
getPartitionSpec is used exclusively when AnalyzePartitionCommand is executed.
|
AnalyzePartitionCommand takes the following when created:
noscan flag (enabled by default) that indicates whether NOSCAN option was used or not
AnalyzeColumnCommand is a logical command for ANALYZE TABLE with FOR COLUMNS clause (and no PARTITION specification).
|
1 2 3 4 5 |
ANALYZE TABLE tableName COMPUTE STATISTICS FOR COLUMNS columnNames |
|
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 50 |
// Make the example reproducible val tableName = "t1" import org.apache.spark.sql.catalyst.TableIdentifier val tableId = TableIdentifier(tableName) val sessionCatalog = spark.sessionState.catalog sessionCatalog.dropTable(tableId, ignoreIfNotExists = true, purge = true) val df = Seq((0, 0.0, "zero"), (1, 1.4, "one")).toDF("id", "p1", "p2") df.write.saveAsTable("t1") // AnalyzeColumnCommand represents ANALYZE TABLE...FOR COLUMNS SQL command val allCols = df.columns.mkString(",") val analyzeTableSQL = s"ANALYZE TABLE $tableName COMPUTE STATISTICS FOR COLUMNS $allCols" val plan = spark.sql(analyzeTableSQL).queryExecution.logical import org.apache.spark.sql.execution.command.AnalyzeColumnCommand val cmd = plan.asInstanceOf[AnalyzeColumnCommand] scala> println(cmd) AnalyzeColumnCommand `t1`, [id, p1, p2] spark.sql(analyzeTableSQL) val stats = sessionCatalog.getTableMetadata(tableId).stats.get scala> println(stats.simpleString) 1421 bytes, 2 rows scala> stats.colStats.map { case (c, ss) => s"$c: $ss" }.foreach(println) id: ColumnStat(2,Some(0),Some(1),0,4,4,None) p1: ColumnStat(2,Some(0.0),Some(1.4),0,8,8,None) p2: ColumnStat(2,None,None,0,4,4,None) // Use DESC EXTENDED for friendlier output scala> sql(s"DESC EXTENDED $tableName id").show +--------------+----------+ | info_name|info_value| +--------------+----------+ | col_name| id| | data_type| int| | comment| NULL| | min| 0| | max| 1| | num_nulls| 0| |distinct_count| 2| | avg_col_len| 4| | max_col_len| 4| | histogram| NULL| +--------------+----------+ |
AnalyzeColumnCommand can generate column histograms when spark.sql.statistics.histogram.enabled configuration property is turned on (which is disabled by default). AnalyzeColumnCommand supports column histograms for the following data types:
IntegralType
DecimalType
DoubleType
FloatType
DateType
TimestampType
|
Note
|
Histograms can provide better estimation accuracy. Currently, Spark only supports equi-height histogram. Note that collecting histograms takes extra cost. For example, collecting column statistics usually takes only one table scan, but generating equi-height histogram will cause an extra table scan. |
|
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 |
// ./bin/spark-shell --conf spark.sql.statistics.histogram.enabled=true // Use the above example to set up the environment // Make sure that ANALYZE TABLE COMPUTE STATISTICS FOR COLUMNS was run with histogram enabled // There are 254 bins by default // Use spark.sql.statistics.histogram.numBins to control the bins val descExtSQL = s"DESC EXTENDED $tableName p1" scala> spark.sql(descExtSQL).show(truncate = false) +--------------+-----------------------------------------------------+ |info_name |info_value | +--------------+-----------------------------------------------------+ |col_name |p1 | |data_type |double | |comment |NULL | |min |0.0 | |max |1.4 | |num_nulls |0 | |distinct_count|2 | |avg_col_len |8 | |max_col_len |8 | |histogram |height: 0.007874015748031496, num_of_bins: 254 | |bin_0 |lower_bound: 0.0, upper_bound: 0.0, distinct_count: 1| |bin_1 |lower_bound: 0.0, upper_bound: 0.0, distinct_count: 1| |bin_2 |lower_bound: 0.0, upper_bound: 0.0, distinct_count: 1| |bin_3 |lower_bound: 0.0, upper_bound: 0.0, distinct_count: 1| |bin_4 |lower_bound: 0.0, upper_bound: 0.0, distinct_count: 1| |bin_5 |lower_bound: 0.0, upper_bound: 0.0, distinct_count: 1| |bin_6 |lower_bound: 0.0, upper_bound: 0.0, distinct_count: 1| |bin_7 |lower_bound: 0.0, upper_bound: 0.0, distinct_count: 1| |bin_8 |lower_bound: 0.0, upper_bound: 0.0, distinct_count: 1| |bin_9 |lower_bound: 0.0, upper_bound: 0.0, distinct_count: 1| +--------------+-----------------------------------------------------+ only showing top 20 rows |
|
Note
|
AnalyzeColumnCommand is described by analyze labeled alternative in statement expression in SqlBase.g4 and parsed using SparkSqlAstBuilder.
|
|
Note
|
AnalyzeColumnCommand is not supported on views.
|
run Method|
1 2 3 4 5 |
run(sparkSession: SparkSession): Seq[Row] |
|
Note
|
run is part of RunnableCommand Contract to execute (run) a logical command.
|
run calculates the following statistics:
sizeInBytes
stats for each column
|
Caution
|
FIXME |
computeColumnStats Internal Method|
1 2 3 4 5 6 7 8 |
computeColumnStats( sparkSession: SparkSession, tableIdent: TableIdentifier, columnNames: Seq[String]): (Long, Map[String, ColumnStat]) |
computeColumnStats…FIXME
|
Note
|
computeColumnStats is used exclusively when AnalyzeColumnCommand is executed.
|
computePercentiles Internal Method|
1 2 3 4 5 6 7 8 |
computePercentiles( attributesToAnalyze: Seq[Attribute], sparkSession: SparkSession, relation: LogicalPlan): AttributeMap[ArrayData] |
computePercentiles…FIXME
|
Note
|
computePercentiles is used exclusively when AnalyzeColumnCommand is executed (and computes column statistics).
|
AnalysisBarrier is a leaf logical operator that is a wrapper of an analyzed logical plan to hide it from the Spark Analyzer. The purpose of AnalysisBarrier is to prevent the child logical plan from being analyzed again (and increasing the time spent on query analysis).
AnalysisBarrier is created when:
ResolveReferences logical resolution rule is requested to dedupRight
ResolveMissingReferences logical resolution rule is requested to resolveExprsAndAddMissingAttrs
Dataset is created
DataFrameWriter is requested to execute a logical command for writing to a data source V1 (when DataFrameWriter is requested to save the rows of a structured query (a DataFrame) to a data source)
KeyValueGroupedDataset is requested for the logical query plan
AnalysisBarrier takes a single child logical query plan when created.
AnalysisBarrier returns the child logical query plan when requested for the inner nodes (that should be shown as an inner nested tree of this node).
AnalysisBarrier simply requests the child logical query plan for the output schema attributes.
AnalysisBarrier simply requests the child logical query plan for the isStreaming flag.
AnalysisBarrier simply requests the child logical operator for the canonicalized version.
AlterViewAsCommand is a logical command for ALTER VIEW SQL statement to alter a view.
AlterViewAsCommand works with a table identifier (as TableIdentifier), the original SQL text, and a LogicalPlan for the SQL query.
|
Note
|
AlterViewAsCommand is described by alterViewQuery labeled alternative in statement expression in SqlBase.g4 and parsed using SparkSqlParser.
|
When executed, AlterViewAsCommand attempts to alter a temporary view in the current SessionCatalog first, and if that “fails”, alters the permanent view.
run Method|
1 2 3 4 5 |
run(session: SparkSession): Seq[Row] |
|
Note
|
run is part of RunnableCommand Contract to execute (run) a logical command.
|
run…FIXME
Aggregate is a unary logical operator that holds the following:
Grouping expressions
Aggregate named expressions
Child logical plan
Aggregate is created to represent the following (after a logical plan is analyzed):
SQL’s GROUP BY clause (possibly with WITH CUBE or WITH ROLLUP)
RelationalGroupedDataset aggregations (e.g. pivot)
KeyValueGroupedDataset aggregations
AnalyzeColumnCommand logical command
|
Note
|
Aggregate logical operator is translated to one of HashAggregateExec, ObjectHashAggregateExec or SortAggregateExec physical operators in Aggregation execution planning strategy.
|
| Name | Description | ||
|---|---|---|---|
|
|
Child logical plan‘s
|
||
|
|
Attributes of aggregate named expressions
|
||
|
|
Enabled when:
|
||
|
|
The (expression) constraints of child logical plan and non-aggregate aggregate named expressions.
|
PushDownPredicate logical plan optimization applies so-called filter pushdown to a Pivot operator when under Filter operator and with all expressions deterministic.
|
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 |
import org.apache.spark.sql.catalyst.optimizer.PushDownPredicate val q = visits .groupBy("city") .pivot("year") .count() .where($"city" === "Boston") val pivotPlanAnalyzed = q.queryExecution.analyzed scala> println(pivotPlanAnalyzed.numberedTreeString) 00 Filter (city#8 = Boston) 01 +- Project [city#8, __pivot_count(1) AS `count` AS `count(1) AS ``count```#142[0] AS 2015#143L, __pivot_count(1) AS `count` AS `count(1) AS ``count```#142[1] AS 2016#144L, __pivot_count(1) AS `count` AS `count(1) AS ``count```#142[2] AS 2017#145L] 02 +- Aggregate [city#8], [city#8, pivotfirst(year#9, count(1) AS `count`#134L, 2015, 2016, 2017, 0, 0) AS __pivot_count(1) AS `count` AS `count(1) AS ``count```#142] 03 +- Aggregate [city#8, year#9], [city#8, year#9, count(1) AS count(1) AS `count`#134L] 04 +- Project [_1#3 AS id#7, _2#4 AS city#8, _3#5 AS year#9] 05 +- LocalRelation [_1#3, _2#4, _3#5] val afterPushDown = PushDownPredicate(pivotPlanAnalyzed) scala> println(afterPushDown.numberedTreeString) 00 Project [city#8, __pivot_count(1) AS `count` AS `count(1) AS ``count```#142[0] AS 2015#143L, __pivot_count(1) AS `count` AS `count(1) AS ``count```#142[1] AS 2016#144L, __pivot_count(1) AS `count` AS `count(1) AS ``count```#142[2] AS 2017#145L] 01 +- Aggregate [city#8], [city#8, pivotfirst(year#9, count(1) AS `count`#134L, 2015, 2016, 2017, 0, 0) AS __pivot_count(1) AS `count` AS `count(1) AS ``count```#142] 02 +- Aggregate [city#8, year#9], [city#8, year#9, count(1) AS count(1) AS `count`#134L] 03 +- Project [_1#3 AS id#7, _2#4 AS city#8, _3#5 AS year#9] 04 +- Filter (_2#4 = Boston) 05 +- LocalRelation [_1#3, _2#4, _3#5] |
SaveAsHiveFile is the extension of the DataWritingCommand contract for commands that saveAsHiveFile.
| SaveAsHiveFile | Description |
|---|---|
saveAsHiveFile Method|
1 2 3 4 5 6 7 8 9 10 11 12 |
saveAsHiveFile( sparkSession: SparkSession, plan: SparkPlan, hadoopConf: Configuration, fileSinkConf: FileSinkDesc, outputLocation: String, customPartitionLocations: Map[TablePartitionSpec, String] = Map.empty, partitionAttributes: Seq[Attribute] = Nil): Set[String] |
saveAsHiveFile…FIXME
|
Note
|
|
DataWritingCommand is an extension of the Command contract for logical commands that write query data to a relation when executed.
DataWritingCommand is resolved to a DataWritingCommandExec physical operator when BasicOperators execution planning strategy is executed (i.e. plan a logical plan to a physical plan).
| Property | Description | ||
|---|---|---|---|
|
|
The output column names of the analyzed input query plan Used when |
||
|
|
The logical query plan representing the data to write (i.e. whose result will be inserted into a relation) Used when |
||
|
|
Executes the command Used when:
|
When requested for the child nodes, DataWritingCommand simply returns the logical query plan.
DataWritingCommand defines custom performance metrics.
| Key | Name (in web UI) | Description |
|---|---|---|
|
|
number of written files |
|
|
|
bytes of written output |
|
|
|
number of output rows |
|
|
|
number of dynamic part |
The performance metrics are used when:
DataWritingCommand is requested for the BasicWriteJobStatsTracker
DataWritingCommandExec physical operator is requested for the metrics
| DataWritingCommand | Description |
|---|---|
basicWriteJobStatsTracker Method|
1 2 3 4 5 |
basicWriteJobStatsTracker(hadoopConf: Configuration): BasicWriteJobStatsTracker |
basicWriteJobStatsTracker simply creates and returns a new BasicWriteJobStatsTracker (with the given Hadoop Configuration and the metrics).
|
Note
|
|
outputColumns Method|
1 2 3 4 5 |
outputColumns: Seq[Attribute] |
outputColumns…FIXME
|
Note
|
|
