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

ColumnarBatch

ColumnarBatch is…FIXME

ColumnarBatch is created when:

InMemoryTableScanExec is requested to createAndDecompressColumn
VectorizedParquetRecordReader is requested to initBatch
OrcColumnarBatchReader is requested to initBatch
ColumnVectorUtils is requested to toBatch
ArrowPythonRunner is requested for a Iterator[ColumnarBatch] (i.e. newReaderIterator)
ArrowConverters is requested for a ArrowRowIterator (i.e. fromPayloadIterator)

Note	`ColumnarBatch` is an `Evolving` contract that is evolving towards becoming a stable API, but is not a stable API yet and can change from one feature release to another release. In other words, using the contract is as treading on thin ice.

ColumnarBatch takes an array of ColumnVectors when created. ColumnarBatch initializes the internal MutableColumnarRow.

Table 1. ColumnarBatch’s Internal Properties (e.g. Registries, Counters and Flags)
Name	Description
`numRows`	Number of rows
`row`	`MutableColumnarRow`

Vectorized Query Execution (Batch Decoding)

2012-04-07admin阅读(1265)

Vectorized Query Execution (Batch Decoding)

Vectorized Query Execution (aka Batch Decoding) is…FIXME

BytesToBytesMap Append-Only Hash Map

2012-04-06admin阅读(1461)

BytesToBytesMap Append-Only Hash Map

BytesToBytesMap is…FIXME

Low space overhead,
Good memory locality, esp. for scans.

`lookup` Method



Location lookup(Object keyBase, long keyOffset, int keyLength)
Location lookup(Object keyBase, long keyOffset, int keyLength, int hash)

Location lookup(Object keyBase, long keyOffset, int keyLength)

Location lookup(Object keyBase, long keyOffset, int keyLength, int hash)

Caution

FIXME

`safeLookup` Method



void safeLookup(Object keyBase, long keyOffset, int keyLength, Location loc, int hash)

void safeLookup(Object keyBase, long keyOffset, int keyLength, Location loc, int hash)

safeLookup…FIXME

Note	`safeLookup` is used when `BytesToBytesMap` does lookup and `UnsafeHashedRelation` for looking up a single value or values by key.

GenerateSafeProjection

GenerateSafeProjection is…FIXME

Creating Projection — `create` Method



create(expressions: Seq[Expression]): Projection

create(expressions: Seq[Expression]): Projection

Note	`create` is part of CodeGenerator Contract to…FIXME.

create…FIXME

GeneratePredicate

GeneratePredicate is…FIXME

Creating Predicate — `create` Method



create(predicate: Expression): Predicate

create(predicate: Expression): Predicate

Note	`create` is part of CodeGenerator Contract to…FIXME.

create…FIXME

GenerateOrdering

GenerateOrdering is…FIXME

Creating BaseOrdering — `create` Method



create(ordering: Seq[SortOrder]): BaseOrdering
create(schema: StructType): BaseOrdering

create(ordering: Seq[SortOrder]): BaseOrdering

create(schema: StructType): BaseOrdering

Note	`create` is part of CodeGenerator Contract to…FIXME.

create…FIXME

`genComparisons` Method



genComparisons(ctx: CodegenContext, schema: StructType): String

genComparisons(ctx: CodegenContext, schema: StructType): String

genComparisons…FIXME

Note	`genComparisons` is used when…FIXME

GenerateColumnAccessor

GenerateColumnAccessor is a CodeGenerator for…FIXME

Creating ColumnarIterator — `create` Method



create(columnTypes: Seq[DataType]): ColumnarIterator

create(columnTypes: Seq[DataType]): ColumnarIterator

Note	`create` is part of CodeGenerator Contract to…FIXME.

create…FIXME

CodeGenerator

CodeGenerator is a base class for generators of JVM bytecode for expression evaluation.

Table 1. CodeGenerator’s Internal Properties
Name	Description
`cache`	Guava’s LoadingCache with at most 100 pairs of `CodeAndComment` and `GeneratedClass`.
`genericMutableRowType`

Tip

Enable INFO or DEBUG logging level for org.apache.spark.sql.catalyst.expressions.codegen.CodeGenerator logger to see what happens inside.

Add the following line to conf/log4j.properties:



log4j.logger.org.apache.spark.sql.catalyst.expressions.codegen.CodeGenerator=DEBUG

log4j.logger.org.apache.spark.sql.catalyst.expressions.codegen.CodeGenerator=DEBUG

Refer to Logging.

CodeGenerator Contract



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

abstract class CodeGenerator[InType, OutType] {
  def create(in: InType): OutType
  def canonicalize(in: InType): InType
  def bind(in: InType, inputSchema: Seq[Attribute]): InType
  def generate(expressions: InType, inputSchema: Seq[Attribute]): OutType
  def generate(expressions: InType): OutType
}

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

abstract class CodeGenerator[InType, OutType] {

def create(in: InType): OutType

def canonicalize(in: InType): InType

def bind(in: InType, inputSchema: Seq[Attribute]): InType

def generate(expressions: InType, inputSchema: Seq[Attribute]): OutType

def generate(expressions: InType): OutType

}

Table 2. CodeGenerator Contract
Method	Description
`generate`	Generates an evaluator for expression(s) that may (optionally) have expression(s) bound to a schema (i.e. a collection of Attribute). Used in: `ExpressionEncoder` for UnsafeProjection (for serialization)

Compiling Java Source Code using Janino — `doCompile` Internal Method

Caution

FIXME

Finding or Compiling Java Source Code — `compile` Method

Caution

FIXME

`create` Method



create(references: Seq[Expression]): UnsafeProjection

create(references: Seq[Expression]): UnsafeProjection

Caution

FIXME

Note	`create` is used when: `CodeGenerator` generates an expression evaluator `GenerateOrdering` creates a code gen ordering for `SortOrder` expressions

Creating CodegenContext — `newCodeGenContext` Method



newCodeGenContext(): CodegenContext

newCodeGenContext(): CodegenContext

newCodeGenContext simply creates a new CodegenContext.

Note

newCodeGenContext is used when:

GenerateMutableProjection is requested to create a MutableProjection
GenerateOrdering is requested to create a BaseOrdering
GeneratePredicate is requested to create a Predicate
GenerateSafeProjection is requested to create a Projection
GenerateUnsafeProjection is requested to create a UnsafeProjection
GenerateColumnAccessor is requested to create a ColumnarIterator

CodegenContext

CodegenContext is…FIXME

CodegenContext takes no input parameters.



import org.apache.spark.sql.catalyst.expressions.codegen.CodegenContext
val ctx = new CodegenContext

import org.apache.spark.sql.catalyst.expressions.codegen.CodegenContext

val ctx = new CodegenContext

CodegenContext is created when:

WholeStageCodegenExec physical operator is requested to generate a Java source code for the child operator (when WholeStageCodegenExec is executed)
CodeGenerator is requested for a new CodegenContext
GenerateUnsafeRowJoiner is requested for a UnsafeRowJoiner

CodegenContext stores expressions that don’t support codegen.


Example of CodegenContext.subexpressionElimination (through CodegenContext.generateExpressions)

import org.apache.spark.sql.catalyst.expressions.codegen.CodegenContext
val ctx = new CodegenContext

// Use Catalyst DSL
import org.apache.spark.sql.catalyst.dsl.expressions._
val expressions = "hello".expr.as("world") :: "hello".expr.as("world") :: Nil

// FIXME Use a real-life query to extract the expressions

// CodegenContext.subexpressionElimination (where the elimination all happens) is a private method
// It is used exclusively in CodegenContext.generateExpressions which is public
// and does the elimination when it is enabled

// Note the doSubexpressionElimination flag is on
// Triggers the subexpressionElimination private method
ctx.generateExpressions(expressions, doSubexpressionElimination = true)

// subexpressionElimination private method uses ctx.equivalentExpressions
val commonExprs = ctx.equivalentExpressions.getAllEquivalentExprs

assert(commonExprs.length > 0, "No common expressions found")

Example of CodegenContext.subexpressionElimination (through CodegenContext.generateExpressions)

import org.apache.spark.sql.catalyst.expressions.codegen.CodegenContext

val ctx = new CodegenContext

// Use Catalyst DSL

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

val expressions = "hello".expr.as("world") :: "hello".expr.as("world") :: Nil

// FIXME Use a real-life query to extract the expressions

// CodegenContext.subexpressionElimination (where the elimination all happens) is a private method

// It is used exclusively in CodegenContext.generateExpressions which is public

// and does the elimination when it is enabled

// Note the doSubexpressionElimination flag is on

// Triggers the subexpressionElimination private method

ctx.generateExpressions(expressions, doSubexpressionElimination = true)

// subexpressionElimination private method uses ctx.equivalentExpressions

val commonExprs = ctx.equivalentExpressions.getAllEquivalentExprs

assert(commonExprs.length > 0, "No common expressions found")

Table 1. CodegenContext’s Internal Properties (e.g. Registries, Counters and Flags)
Name	Description
`classFunctions`	Mutable Scala `Map` with function names, their Java source code and a class name New entries are added when `CodegenContext` is requested to addClass and addNewFunctionToClass Used when `CodegenContext` is requested to declareAddedFunctions
`equivalentExpressions`	EquivalentExpressions Expressions are added and then fetched as equivalent sets when `CodegenContext` is requested to subexpressionElimination (for generateExpressions with subexpression elimination enabled)
`currentVars`	The list of generated columns as input of current operator
`INPUT_ROW`	The variable name of the input row of the current operator
`placeHolderToComments`	Placeholders and their comments Used when…FIXME
`references`	References that are used to generate classes in the following code generators: GenerateMutableProjection GenerateOrdering GeneratePredicate GenerateSafeProjection GenerateUnsafeProjection WholeStageCodegenExec Elements are added when: `CodegenContext` is requested to addReferenceObj `CodegenFallback` is requested to doGenCode
`subExprEliminationExprs`	`SubExprEliminationStates` by Expression Used when…FIXME
`subexprFunctions`	Names of the functions that…FIXME

Generating Java Source Code For Code-Generated Evaluation of Multiple Expressions (With Optional Subexpression Elimination) — `generateExpressions` Method



generateExpressions(
  expressions: Seq[Expression],
  doSubexpressionElimination: Boolean = false): Seq[ExprCode]

generateExpressions(

expressions: Seq[Expression],

doSubexpressionElimination: Boolean = false): Seq[ExprCode]

(only with subexpression elimination enabled) generateExpressions does subexpressionElimination of the input expressions.

In the end, generateExpressions requests every expressions to generate the Java source code for code-generated (non-interpreted) expression evaluation.

Note

generateExpressions is used when:

GenerateMutableProjection is requested to create a MutableProjection
GenerateUnsafeProjection is requested to create an ExprCode for Catalyst expressions
HashAggregateExec is requested to generate the Java source code for whole-stage consume path with grouping keys

`addReferenceObj` Method



addReferenceObj(objName: String, obj: Any, className: String = null): String

addReferenceObj(objName: String, obj: Any, className: String = null): String

addReferenceObj…FIXME

Note	`addReferenceObj` is used when…FIXME

`subexpressionEliminationForWholeStageCodegen` Method



subexpressionEliminationForWholeStageCodegen(expressions: Seq[Expression]): SubExprCodes

subexpressionEliminationForWholeStageCodegen(expressions: Seq[Expression]): SubExprCodes

subexpressionEliminationForWholeStageCodegen…FIXME

Note	`subexpressionEliminationForWholeStageCodegen` is used exclusively when `HashAggregateExec` is requested to generate a Java source code for whole-stage consume path (with grouping keys or not).

Adding Function to Generated Class — `addNewFunction` Method



addNewFunction(
  funcName: String,
  funcCode: String,
  inlineToOuterClass: Boolean = false): String

addNewFunction(

funcName: String,

funcCode: String,

inlineToOuterClass: Boolean = false): String

addNewFunction…FIXME

Note	`addNewFunction` is used when…FIXME

`subexpressionElimination` Internal Method



subexpressionElimination(expressions: Seq[Expression]): Unit

subexpressionElimination(expressions: Seq[Expression]): Unit

subexpressionElimination requests EquivalentExpressions to addExprTree for every expression (in the input expressions).

subexpressionElimination requests EquivalentExpressions for the equivalent sets of expressions with at least two equivalent expressions (aka common expressions).

For every equivalent expression set, subexpressionElimination does the following:

Takes the first expression and requests it to generate a Java source code for the expression tree
addNewFunction and adds it to subexprFunctions
Creates a SubExprEliminationState and adds it with every common expression in the equivalent expression set to subExprEliminationExprs

Note	`subexpressionElimination` is used exclusively when `CodegenContext` is requested to generateExpressions (with subexpression elimination enabled).

Adding Mutable State — `addMutableState` Method



addMutableState(
  javaType: String,
  variableName: String,
  initFunc: String => String = _ => "",
  forceInline: Boolean = false,
  useFreshName: Boolean = true): String

addMutableState(

javaType: String,

variableName: String,

initFunc: String => String = _ => "",

forceInline: Boolean = false,

useFreshName: Boolean = true): String

addMutableState…FIXME



val input = ctx.addMutableState("scala.collection.Iterator", "input", v => s"$v = inputs[0];")

val input = ctx.addMutableState("scala.collection.Iterator", "input", v => s"$v = inputs[0];")

Note	`addMutableState` is used when…FIXME

Adding Immutable State (Unless Exists Already) — `addImmutableStateIfNotExists` Method



addImmutableStateIfNotExists(
  javaType: String,
  variableName: String,
  initFunc: String => String = _ => ""): Unit

addImmutableStateIfNotExists(

javaType: String,

variableName: String,

initFunc: String => String = _ => ""): Unit

addImmutableStateIfNotExists…FIXME



val ctx: CodegenContext = ???
val partitionMaskTerm = "partitionMask"
ctx.addImmutableStateIfNotExists(ctx.JAVA_LONG, partitionMaskTerm)

val ctx: CodegenContext = ???

val partitionMaskTerm = "partitionMask"

ctx.addImmutableStateIfNotExists(ctx.JAVA_LONG, partitionMaskTerm)

Note	`addImmutableStateIfNotExists` is used when…FIXME

`freshName` Method



freshName(name: String): String

freshName(name: String): String

freshName…FIXME

Note	`freshName` is used when…FIXME

`addNewFunctionToClass` Internal Method



addNewFunctionToClass(
  funcName: String,
  funcCode: String,
  className: String): mutable.Map[String, mutable.Map[String, String]]

addNewFunctionToClass(

funcName: String,

funcCode: String,

className: String): mutable.Map[String, mutable.Map[String, String]]

addNewFunctionToClass…FIXME

Note	`addNewFunctionToClass` is used when…FIXME

`addClass` Internal Method



addClass(className: String, classInstance: String): Unit

addClass(className: String, classInstance: String): Unit

addClass…FIXME

Note	`addClass` is used when…FIXME

`declareAddedFunctions` Method



declareAddedFunctions(): String

declareAddedFunctions(): String

declareAddedFunctions…FIXME

Note	`declareAddedFunctions` is used when…FIXME

`declareMutableStates` Method



declareMutableStates(): String

declareMutableStates(): String

declareMutableStates…FIXME

Note	`declareMutableStates` is used when…FIXME

`initMutableStates` Method



initMutableStates(): String

initMutableStates(): String

initMutableStates…FIXME

Note	`initMutableStates` is used when…FIXME

`initPartition` Method



initPartition(): String

initPartition(): String

initPartition…FIXME

Note	`initPartition` is used when…FIXME

`emitExtraCode` Method



emitExtraCode(): String

emitExtraCode(): String

emitExtraCode…FIXME

Note	`emitExtraCode` is used when…FIXME

`addPartitionInitializationStatement` Method



addPartitionInitializationStatement(statement: String): Unit

addPartitionInitializationStatement(statement: String): Unit

addPartitionInitializationStatement…FIXME

Note	`addPartitionInitializationStatement` is used when…FIXME

Whole-Stage Java Code Generation (Whole-Stage CodeGen)

2012-03-30admin阅读(2065)

Whole-Stage Java Code Generation (Whole-Stage CodeGen)

Whole-Stage Java Code Generation (aka Whole-Stage CodeGen) is a physical query optimization in Spark SQL that fuses multiple physical operators (as a subtree of plans that support code generation) together into a single Java function.

Whole-Stage Java Code Generation improves the execution performance of a query by collapsing a query tree into a single optimized function that eliminates virtual function calls and leverages CPU registers for intermediate data.

Note

Whole-Stage Code Generation is controlled by spark.sql.codegen.wholeStage Spark internal property.

Whole-Stage Code Generation is enabled by default.



import org.apache.spark.sql.internal.SQLConf.WHOLESTAGE_CODEGEN_ENABLED
scala> spark.conf.get(WHOLESTAGE_CODEGEN_ENABLED)
res0: String = true

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

scala> spark.conf.get(WHOLESTAGE_CODEGEN_ENABLED)

res0: String = true

Use SQLConf.wholeStageEnabled method to access the current value.



scala> spark.sessionState.conf.wholeStageEnabled
res1: Boolean = true

scala> spark.sessionState.conf.wholeStageEnabled

res1: Boolean = true

Note	Whole-Stage Code Generation is used by some modern massively parallel processing (MPP) databases to achieve a better query execution performance. See Efficiently Compiling Efficient Query Plans for Modern Hardware (PDF).

Note	Janino is used to compile a Java source code into a Java class at runtime.

Before a query is executed, CollapseCodegenStages physical preparation rule finds the physical query plans that support codegen and collapses them together as WholeStageCodegen (possibly with InputAdapter in-between for physical operators with no support for Java code generation).

Note	`CollapseCodegenStages` is part of the sequence of physical preparation rules QueryExecution.preparations that will be applied in order to the physical plan before execution.

There are the following code generation paths (as coined in this commit):

Non-whole-stage-codegen path

Whole-stage-codegen “produce” path

Whole-stage-codegen “consume” path

Tip	Review SPARK-12795 Whole stage codegen to learn about the work to support it.

BenchmarkWholeStageCodegen — Performance Benchmark

BenchmarkWholeStageCodegen class provides a benchmark to measure whole stage codegen performance.

You can execute it using the command:



build/sbt 'sql/testOnly *BenchmarkWholeStageCodegen'

build/sbt 'sql/testOnly *BenchmarkWholeStageCodegen'

Note	You need to un-ignore tests in `BenchmarkWholeStageCodegen` by replacing `ignore` with `test`.



$ build/sbt 'sql/testOnly *BenchmarkWholeStageCodegen'
...
Running benchmark: range/limit/sum
  Running case: range/limit/sum codegen=false
22:55:23.028 WARN org.apache.hadoop.util.NativeCodeLoader: Unable to load native-hadoop library for your platform... using builtin-java classes where applicable
  Running case: range/limit/sum codegen=true

Java HotSpot(TM) 64-Bit Server VM 1.8.0_77-b03 on Mac OS X 10.10.5
Intel(R) Core(TM) i7-4870HQ CPU @ 2.50GHz

range/limit/sum:                    Best/Avg Time(ms)    Rate(M/s)   Per Row(ns)   Relative
-------------------------------------------------------------------------------------------
range/limit/sum codegen=false             376 /  433       1394.5           0.7       1.0X
range/limit/sum codegen=true              332 /  388       1581.3           0.6       1.1X

[info] - range/limit/sum (10 seconds, 74 milliseconds)

$ build/sbt 'sql/testOnly *BenchmarkWholeStageCodegen'

...

Running benchmark: range/limit/sum

Running case: range/limit/sum codegen=false

22:55:23.028 WARN org.apache.hadoop.util.NativeCodeLoader: Unable to load native-hadoop library for your platform... using builtin-java classes where applicable

Running case: range/limit/sum codegen=true

Java HotSpot(TM) 64-Bit Server VM 1.8.0_77-b03 on Mac OS X 10.10.5

Intel(R) Core(TM) i7-4870HQ CPU @ 2.50GHz

range/limit/sum: Best/Avg Time(ms) Rate(M/s) Per Row(ns) Relative

-------------------------------------------------------------------------------------------

range/limit/sum codegen=false 376 / 433 1394.5 0.7 1.0X

range/limit/sum codegen=true 332 / 388 1581.3 0.6 1.1X

[info] - range/limit/sum (10 seconds, 74 milliseconds)

上一页
1
···
46
47
48
49
50
51
52
...
下一页
共 58 页

spark-sql 第49页

ColumnarBatch

Vectorized Query Execution (Batch Decoding)

BytesToBytesMap Append-Only Hash Map

lookup Method

safeLookup Method

GenerateSafeProjection

Creating Projection — create Method

GeneratePredicate

Creating Predicate — create Method

GenerateOrdering

Creating BaseOrdering — create Method

genComparisons Method

GenerateColumnAccessor

Creating ColumnarIterator — create Method

CodeGenerator

CodeGenerator Contract

Compiling Java Source Code using Janino — doCompile Internal Method

Finding or Compiling Java Source Code — compile Method

create Method

Creating CodegenContext — newCodeGenContext Method

CodegenContext

Generating Java Source Code For Code-Generated Evaluation of Multiple Expressions (With Optional Subexpression Elimination) — generateExpressions Method

addReferenceObj Method

subexpressionEliminationForWholeStageCodegen Method

Adding Function to Generated Class — addNewFunction Method

subexpressionElimination Internal Method

Adding Mutable State — addMutableState Method

Adding Immutable State (Unless Exists Already) — addImmutableStateIfNotExists Method

freshName Method

addNewFunctionToClass Internal Method

addClass Internal Method

declareAddedFunctions Method

declareMutableStates Method

initMutableStates Method

initPartition Method

emitExtraCode Method

addPartitionInitializationStatement Method

Whole-Stage Java Code Generation (Whole-Stage CodeGen)

BenchmarkWholeStageCodegen — Performance Benchmark

欢迎关注：spark技术分享

关注公众号：spark技术分享

QQ咨询

回顶部

`lookup` Method

`safeLookup` Method

Creating Projection — `create` Method

Creating Predicate — `create` Method

Creating BaseOrdering — `create` Method

`genComparisons` Method

Creating ColumnarIterator — `create` Method

Compiling Java Source Code using Janino — `doCompile` Internal Method

Finding or Compiling Java Source Code — `compile` Method

`create` Method

Creating CodegenContext — `newCodeGenContext` Method

Generating Java Source Code For Code-Generated Evaluation of Multiple Expressions (With Optional Subexpression Elimination) — `generateExpressions` Method

`addReferenceObj` Method

`subexpressionEliminationForWholeStageCodegen` Method

Adding Function to Generated Class — `addNewFunction` Method

`subexpressionElimination` Internal Method

Adding Mutable State — `addMutableState` Method

Adding Immutable State (Unless Exists Already) — `addImmutableStateIfNotExists` Method

`freshName` Method

`addNewFunctionToClass` Internal Method

`addClass` Internal Method

`declareAddedFunctions` Method

`declareMutableStates` Method

`initMutableStates` Method

`initPartition` Method

`emitExtraCode` Method

`addPartitionInitializationStatement` Method