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

BroadcastHashJoinExec Binary Physical Operator for Broadcast Hash Join

BroadcastHashJoinExec is a binary physical operator to perform a broadcast hash join.

BroadcastHashJoinExec is created after applying JoinSelection execution planning strategy to ExtractEquiJoinKeys-destructurable logical query plans (i.e. INNER, CROSS, LEFT OUTER, LEFT SEMI, LEFT ANTI) of which the right physical operator can be broadcast.

BroadcastHashJoinExec supports Java code generation (aka codegen).



val tokens = Seq(
  (0, "playing"),
  (1, "with"),
  (2, "BroadcastHashJoinExec")
).toDF("id", "token")

scala> spark.conf.get("spark.sql.autoBroadcastJoinThreshold")
res0: String = 10485760

val q = tokens.join(tokens, Seq("id"), "inner")
scala> q.explain
== Physical Plan ==
*Project [id#15, token#16, token#21]
+- *BroadcastHashJoin [id#15], [id#20], Inner, BuildRight
   :- LocalTableScan [id#15, token#16]
   +- BroadcastExchange HashedRelationBroadcastMode(List(cast(input[0, int, false] as bigint)))
      +- LocalTableScan [id#20, token#21]

1

2

3

4

5

6

7

8

9

10

11

12

13

14

15

16

17

18

19

20

21

val tokens = Seq(

(0, "playing"),

(1, "with"),

(2, "BroadcastHashJoinExec")

).toDF("id", "token")

scala> spark.conf.get("spark.sql.autoBroadcastJoinThreshold")

res0: String = 10485760

val q = tokens.join(tokens, Seq("id"), "inner")

scala> q.explain

== Physical Plan ==

*Project [id#15, token#16, token#21]

+- *BroadcastHashJoin [id#15], [id#20], Inner, BuildRight

:- LocalTableScan [id#15, token#16]

+- BroadcastExchange HashedRelationBroadcastMode(List(cast(input[0, int, false] as bigint)))

+- LocalTableScan [id#20, token#21]

BroadcastHashJoinExec requires that partition requirements for the two children physical operators match BroadcastDistribution (with a HashedRelationBroadcastMode) and UnspecifiedDistribution (for left and right sides of a join or vice versa).

Table 1. BroadcastHashJoinExec’s Performance Metrics
Key	Name (in web UI)	Description
`numOutputRows`	number of output rows
`avgHashProbe`	avg hash probe

spark sql BroadcastHashJoinExec webui query details.png

Figure 1. BroadcastHashJoinExec in web UI (Details for Query)

Note	The prefix for variable names for `BroadcastHashJoinExec` operators in CodegenSupport-generated code is bhj.



scala> q.queryExecution.debug.codegen
Found 1 WholeStageCodegen subtrees.
== Subtree 1 / 1 ==
*Project [id#15, token#16, token#21]
+- *BroadcastHashJoin [id#15], [id#20], Inner, BuildRight
   :- LocalTableScan [id#15, token#16]
   +- BroadcastExchange HashedRelationBroadcastMode(List(cast(input[0, int, false] as bigint)))
      +- LocalTableScan [id#20, token#21]

Generated code:
/* 001 */ public Object generate(Object[] references) {
/* 002 */   return new GeneratedIterator(references);
/* 003 */ }
/* 004 */
/* 005 */ final class GeneratedIterator extends org.apache.spark.sql.execution.BufferedRowIterator {
/* 006 */   private Object[] references;
/* 007 */   private scala.collection.Iterator[] inputs;
/* 008 */   private scala.collection.Iterator inputadapter_input;
/* 009 */   private org.apache.spark.broadcast.TorrentBroadcast bhj_broadcast;
/* 010 */   private org.apache.spark.sql.execution.joins.LongHashedRelation bhj_relation;
/* 011 */   private org.apache.spark.sql.execution.metric.SQLMetric bhj_numOutputRows;
/* 012 */   private UnsafeRow bhj_result;
/* 013 */   private org.apache.spark.sql.catalyst.expressions.codegen.BufferHolder bhj_holder;
/* 014 */   private org.apache.spark.sql.catalyst.expressions.codegen.UnsafeRowWriter bhj_rowWriter;
...

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

scala> q.queryExecution.debug.codegen

Found 1 WholeStageCodegen subtrees.

== Subtree 1 / 1 ==

*Project [id#15, token#16, token#21]

+- *BroadcastHashJoin [id#15], [id#20], Inner, BuildRight

:- LocalTableScan [id#15, token#16]

+- BroadcastExchange HashedRelationBroadcastMode(List(cast(input[0, int, false] as bigint)))

+- LocalTableScan [id#20, token#21]

Generated code:

/* 001 */ public Object generate(Object[] references) {

/* 002 */ return new GeneratedIterator(references);

/* 003 */ }

/* 004 */

/* 005 */ final class GeneratedIterator extends org.apache.spark.sql.execution.BufferedRowIterator {

/* 006 */ private Object[] references;

/* 007 */ private scala.collection.Iterator[] inputs;

/* 008 */ private scala.collection.Iterator inputadapter_input;

/* 009 */ private org.apache.spark.broadcast.TorrentBroadcast bhj_broadcast;

/* 010 */ private org.apache.spark.sql.execution.joins.LongHashedRelation bhj_relation;

/* 011 */ private org.apache.spark.sql.execution.metric.SQLMetric bhj_numOutputRows;

/* 012 */ private UnsafeRow bhj_result;

/* 013 */ private org.apache.spark.sql.catalyst.expressions.codegen.BufferHolder bhj_holder;

/* 014 */ private org.apache.spark.sql.catalyst.expressions.codegen.UnsafeRowWriter bhj_rowWriter;

...

Table 2. BroadcastHashJoinExec’s Required Child Output Distributions
BuildSide	Left Child	Right Child
`BuildLeft`	BroadcastDistribution with HashedRelationBroadcastMode broadcast mode of build join keys	UnspecifiedDistribution
`BuildRight`	UnspecifiedDistribution	BroadcastDistribution with HashedRelationBroadcastMode broadcast mode of build join keys

Executing Physical Operator (Generating RDD[InternalRow]) — `doExecute` Method



doExecute(): RDD[InternalRow]

1

2

3

4

5

doExecute(): RDD[InternalRow]

Note	`doExecute` is part of SparkPlan Contract to generate the runtime representation of a structured query as a distributed computation over internal binary rows on Apache Spark (i.e. `RDD[InternalRow]`).

doExecute…FIXME

Generating Java Source Code for Inner Join — `codegenInner` Internal Method



codegenInner(ctx: CodegenContext, input: Seq[ExprCode]): String

1

2

3

4

5

codegenInner(ctx: CodegenContext, input: Seq[ExprCode]): String

codegenInner…FIXME

Note	`codegenInner` is used exclusively when `BroadcastHashJoinExec` is requested to generate the Java code for the “consume” path in whole-stage code generation.

Generating Java Source Code for Left or Right Outer Join — `codegenOuter` Internal Method



codegenOuter(ctx: CodegenContext, input: Seq[ExprCode]): String

1

2

3

4

5

codegenOuter(ctx: CodegenContext, input: Seq[ExprCode]): String

codegenOuter…FIXME

Note	`codegenOuter` is used exclusively when `BroadcastHashJoinExec` is requested to generate the Java code for the “consume” path in whole-stage code generation.

Generating Java Source Code for Left Semi Join — `codegenSemi` Internal Method



codegenSemi(ctx: CodegenContext, input: Seq[ExprCode]): String

1

2

3

4

5

codegenSemi(ctx: CodegenContext, input: Seq[ExprCode]): String

codegenSemi…FIXME

Note	`codegenSemi` is used exclusively when `BroadcastHashJoinExec` is requested to generate the Java code for the “consume” path in whole-stage code generation.

Generating Java Source Code for Anti Join — `codegenAnti` Internal Method



codegenAnti(ctx: CodegenContext, input: Seq[ExprCode]): String

1

2

3

4

5

codegenAnti(ctx: CodegenContext, input: Seq[ExprCode]): String

codegenAnti…FIXME

Note	`codegenAnti` is used exclusively when `BroadcastHashJoinExec` is requested to generate the Java code for the “consume” path in whole-stage code generation.

`codegenExistence` Internal Method



codegenExistence(ctx: CodegenContext, input: Seq[ExprCode]): String

1

2

3

4

5

codegenExistence(ctx: CodegenContext, input: Seq[ExprCode]): String

codegenExistence…FIXME

Note	`codegenExistence` is used exclusively when `BroadcastHashJoinExec` is requested to generate the Java code for the “consume” path in whole-stage code generation.

`genStreamSideJoinKey` Internal Method



genStreamSideJoinKey(
  ctx: CodegenContext,
  input: Seq[ExprCode]): (ExprCode, String)

1

2

3

4

5

6

7

genStreamSideJoinKey(

ctx: CodegenContext,

input: Seq[ExprCode]): (ExprCode, String)

genStreamSideJoinKey…FIXME

Note	`genStreamSideJoinKey` is used when `BroadcastHashJoinExec` is requested to generate the Java source code for inner, outer, left semi, anti and existence joins (for the “consume” path in whole-stage code generation).

Creating BroadcastHashJoinExec Instance

BroadcastHashJoinExec takes the following when created:

Left join key expressions
Right join key expressions
Join type
BuildSide
Optional join condition expression
Left physical operator
Right physical operator

BroadcastExchangeExec Unary Physical Operator for Broadcast Joins

BroadcastExchangeExec is a Exchange unary physical operator to collect and broadcast rows of a child relation (to worker nodes).

BroadcastExchangeExec is created exclusively when EnsureRequirements physical query plan optimization ensures BroadcastDistribution of the input data of a physical operator (that can really be either BroadcastHashJoinExec or BroadcastNestedLoopJoinExec operators).



val t1 = spark.range(5)
val t2 = spark.range(5)
val q = t1.join(t2).where(t1("id") === t2("id"))

scala> q.explain
== Physical Plan ==
*BroadcastHashJoin [id#19L], [id#22L], Inner, BuildRight
:- *Range (0, 5, step=1, splits=Some(8))
+- BroadcastExchange HashedRelationBroadcastMode(List(input[0, bigint, false]))
   +- *Range (0, 5, step=1, splits=Some(8))

1

2

3

4

5

6

7

8

9

10

11

12

13

14

val t1 = spark.range(5)

val t2 = spark.range(5)

val q = t1.join(t2).where(t1("id") === t2("id"))

scala> q.explain

== Physical Plan ==

*BroadcastHashJoin [id#19L], [id#22L], Inner, BuildRight

:- *Range (0, 5, step=1, splits=Some(8))

+- BroadcastExchange HashedRelationBroadcastMode(List(input[0, bigint, false]))

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

Table 1. BroadcastExchangeExec’s Performance Metrics
Key	Name (in web UI)	Description
`broadcastTime`	time to broadcast (ms)
`buildTime`	time to build (ms)
`collectTime`	time to collect (ms)
`dataSize`	data size (bytes)

spark sql BroadcastExchangeExec webui details for query.png

Figure 1. BroadcastExchangeExec in web UI (Details for Query)

BroadcastExchangeExec uses BroadcastPartitioning partitioning scheme (with the input BroadcastMode).

Waiting Until Relation Has Been Broadcast — `doExecuteBroadcast` Method



def doExecuteBroadcast[T](): broadcast.Broadcast[T]

1

2

3

4

5

def doExecuteBroadcast[T](): broadcast.Broadcast[T]

doExecuteBroadcast waits until the rows are broadcast.

Note	`doExecuteBroadcast` waits spark.sql.broadcastTimeout (defaults to 5 minutes).

Note	`doExecuteBroadcast` is part of SparkPlan Contract to return the result of a structured query as a broadcast variable.

Lazily-Once-Initialized Asynchronously-Broadcast `relationFuture` Internal Attribute



relationFuture: Future[broadcast.Broadcast[Any]]

1

2

3

4

5

relationFuture: Future[broadcast.Broadcast[Any]]

When “materialized” (aka executed), relationFuture finds the current execution id and sets it to the Future thread.

relationFuture requests child physical operator to executeCollectIterator.

relationFuture records the time for executeCollectIterator in collectTime metrics.

Note	`relationFuture` accepts a relation with up to 512 millions rows and 8GB in size, and reports a `SparkException` if the conditions are violated.

relationFuture requests the input BroadcastMode to transform the internal rows to create a relation, e.g. HashedRelation or a Array[InternalRow].

relationFuture calculates the data size:

For a HashedRelation, relationFuture requests it to estimatedSize
For a Array[InternalRow], relationFuture transforms the InternalRows to UnsafeRows and requests each to getSizeInBytes that it sums all up.

relationFuture records the data size as the dataSize metric.

relationFuture records the buildTime metric.

relationFuture requests the SparkContext to broadcast the relation and records the time in broadcastTime metrics.

In the end, relationFuture requests SQLMetrics to post a SparkListenerDriverAccumUpdates (with the execution id and the SQL metrics) and returns the broadcast internal rows.

Note	Since initialization of `relationFuture` happens on the driver, posting a SparkListenerDriverAccumUpdates is the only way how all the SQL metrics could be accessible to other subsystems using `SparkListener` listeners (incl. web UI).

In case of OutOfMemoryError, relationFuture reports another OutOfMemoryError with the following message:



Not enough memory to build and broadcast the table to all worker nodes. As a workaround, you can either disable broadcast by setting spark.sql.autoBroadcastJoinThreshold to -1 or increase the spark driver memory by setting spark.driver.memory to a higher value

1

2

3

4

5

Not enough memory to build and broadcast the table to all worker nodes. As a workaround, you can either disable broadcast by setting spark.sql.autoBroadcastJoinThreshold to -1 or increase the spark driver memory by setting spark.driver.memory to a higher value

Note	`relationFuture` is executed on a separate thread from a custom scala.concurrent.ExecutionContext (built from a cached java.util.concurrent.ThreadPoolExecutor with the prefix broadcast-exchange and up to 128 threads).

Note	`relationFuture` is used when `BroadcastExchangeExec` is requested to prepare for execution (that triggers asynchronous execution of the child operator and broadcasting the result) and execute broadcast (that waits until the broadcasting has finished).

Broadcasting Relation (Rows) Asynchronously — `doPrepare` Method



doPrepare(): Unit

1

2

3

4

5

doPrepare(): Unit

Note	`doPrepare` is part of SparkPlan Contract to prepare a physical operator for execution.

doPrepare simply “materializes” the internal lazily-once-initialized asynchronous broadcast.

Creating BroadcastExchangeExec Instance

BroadcastExchangeExec takes the following when created:

BroadcastMode
Child logical plan

SQLMetric — SQL Execution Metric of Physical Operator

2013-01-01admin阅读(1522)

SQLMetric — SQL Execution Metric of Physical Operator

SQLMetric is a SQL metric for monitoring execution of a physical operator.

SQLMetric is an accumulator (and that is the mechanism to propagate SQL metric updates on the executors to the driver, e.g. web UI).

Note	Use Details for Query page in SQL tab in web UI to see the SQL execution metrics of a structured query.

Note

SQL metrics are collected using SparkListener. If there are no tasks, Spark SQL cannot collect any metrics. Updates to metrics on the driver-side require explicit call of SQLMetrics.postDriverMetricUpdates.

This is why executing some physical operators (e.g. LocalTableScanExec) may not have SQL metrics in web UI’s Details for Query in SQL tab.

Compare the following SQL queries and their execution pages.



// The query does not have SQL metrics in web UI
Seq("Jacek").toDF("name").show

// The query gives numOutputRows metric in web UI's Details for Query (SQL tab)
Seq("Jacek").toDF("name").count

1

2

3

4

5

6

7

8

9

// The query does not have SQL metrics in web UI

Seq("Jacek").toDF("name").show

// The query gives numOutputRows metric in web UI's Details for Query (SQL tab)

Seq("Jacek").toDF("name").count

SQLMetric takes a metric type and an initial value when created.

Table 1. Metric Types and Corresponding Create Methods
Metric Type	Create Method	Failed Values Counted?	Description
`size`	`createSizeMetric`	no	Used when…
`sum`	`createMetric`	no	Used when…
`timing`	`createTimingMetric`	no	Used when…

`reset` Method



reset(): Unit

1

2

3

4

5

reset(): Unit

reset…FIXME

Note	`reset` is used when…FIXME

Posting Driver-Side Metric Updates — `SQLMetrics.postDriverMetricUpdates` Method



postDriverMetricUpdates(
  sc: SparkContext,
  executionId: String,
  metrics: Seq[SQLMetric]): Unit

1

2

3

4

5

6

7

8

postDriverMetricUpdates(

sc: SparkContext,

executionId: String,

metrics: Seq[SQLMetric]): Unit

postDriverMetricUpdates posts a SparkListenerDriverAccumUpdates event to LiveListenerBus when executionId is specified.

Note	`postDriverMetricUpdates` method belongs to `SQLMetrics` object.

Note

postDriverMetricUpdates is used when:

BroadcastExchangeExec is requested to prepare for execution (and initializes relationFuture for the first time)
FileSourceScanExec physical operator is requested for selectedPartitions (and posts updates to numFiles and metadataTime metrics)
SubqueryExec physical operator is requested to prepare for execution (and initializes relationFuture for the first time that in turn posts updates to collectTime and dataSize metrics)

CodeGeneratorWithInterpretedFallback

2012-12-31admin阅读(1253)

CodeGeneratorWithInterpretedFallback

CodeGeneratorWithInterpretedFallback is the base of codegen object generators that can create objects for codegen and interpreted evaluation paths.

Method Description

createCodeGeneratedObject



createCodeGeneratedObject(in: IN): OUT

1

2

3

4

5

createCodeGeneratedObject(in: IN): OUT

Used when…FIXME

createInterpretedObject



createInterpretedObject(in: IN): OUT

1

2

3

4

5

createInterpretedObject(in: IN): OUT

Used when…FIXME

Note	UnsafeProjection is the one and only known implementation of the CodeGeneratorWithInterpretedFallback Contract in Apache Spark.

Note

CodeGeneratorWithInterpretedFallback is a Scala type constructor (aka generic type) that accepts two types referred as IN and OUT.



abstract class CodeGeneratorWithInterpretedFallback[IN, OUT]

1

2

3

4

5

abstract class CodeGeneratorWithInterpretedFallback[IN, OUT]

`createObject` Method



createObject(in: IN): OUT

1

2

3

4

5

createObject(in: IN): OUT

createObject…FIXME

Note	`createObject` is used exclusively when `UnsafeProjection` is requested to create an UnsafeProjection for Catalyst expressions.

InterpretedProjection

2012-12-30admin阅读(1382)

InterpretedProjection

InterpretedProjection is a Projection that…FIXME

InterpretedProjection takes expressions when created.



// HACK: Disable symbolToColumn implicit conversion
// It is imported automatically in spark-shell (and makes demos impossible)
// implicit def symbolToColumn(s: Symbol): org.apache.spark.sql.ColumnName
trait ThatWasABadIdea
implicit def symbolToColumn(ack: ThatWasABadIdea) = ack

import org.apache.spark.sql.catalyst.dsl.expressions._
val boundRef = 'hello.string.at(4)

import org.apache.spark.sql.catalyst.expressions.{Expression, Literal}
val expressions: Seq[Expression] = Seq(Literal(1), boundRef)

import org.apache.spark.sql.catalyst.expressions.InterpretedProjection
val ip = new InterpretedProjection(expressions)
scala> println(ip)
Row => [1,input[4, string, true]]

1

2

3

4

5

6

7

8

9

10

11

12

13

14

15

16

17

18

19

20

// HACK: Disable symbolToColumn implicit conversion

// It is imported automatically in spark-shell (and makes demos impossible)

// implicit def symbolToColumn(s: Symbol): org.apache.spark.sql.ColumnName

trait ThatWasABadIdea

implicit def symbolToColumn(ack: ThatWasABadIdea) = ack

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

val boundRef = 'hello.string.at(4)

import org.apache.spark.sql.catalyst.expressions.{Expression, Literal}

val expressions: Seq[Expression] = Seq(Literal(1), boundRef)

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

val ip = new InterpretedProjection(expressions)

scala> println(ip)

Row => [1,input[4, string, true]]

InterpretedProjection is created when:

UserDefinedGenerator is requested to initializeConverters
ConvertToLocalRelation logical optimization is executed (to transform Project logical operators)
HiveGenericUDTF is evaluated
ScriptTransformationExec physical operator is executed

Initializing Nondeterministic Expressions — `initialize` Method



initialize(partitionIndex: Int): Unit

1

2

3

4

5

initialize(partitionIndex: Int): Unit

Note	`initialize` is part of Projection Contract to…FIXME.

initialize requests Nondeterministic expressions (in expressions) to initialize with the partitionIndex.

GenerateMutableProjection

2012-12-29admin阅读(1415)

GenerateMutableProjection

GenerateMutableProjection is…FIXME

Creating MutableProjection — `create` Internal Method



create(
  expressions: Seq[Expression],
  useSubexprElimination: Boolean): MutableProjection

1

2

3

4

5

6

7

create(

expressions: Seq[Expression],

useSubexprElimination: Boolean): MutableProjection

create…FIXME

Note	`create` is used when…FIXME

GenerateUnsafeProjection

2012-12-28admin阅读(1592)

GenerateUnsafeProjection

GenerateUnsafeProjection is a CodeGenerator that generates the bytecode for a UnsafeProjection for given expressions (i.e. CodeGenerator[Seq[Expression], UnsafeProjection]).



GenerateUnsafeProjection: Seq[Expression] => UnsafeProjection

1

2

3

4

5

GenerateUnsafeProjection: Seq[Expression] => UnsafeProjection

Tip

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

Add the following line to conf/log4j.properties:



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

1

2

3

4

5

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

Refer to Logging.

Generating UnsafeProjection — `generate` Method



generate(
  expressions: Seq[Expression],
  subexpressionEliminationEnabled: Boolean): UnsafeProjection

1

2

3

4

5

6

7

generate(

expressions: Seq[Expression],

subexpressionEliminationEnabled: Boolean): UnsafeProjection

generate canonicalize the input expressions followed by generating a JVM bytecode for a UnsafeProjection for the expressions.

Note

generate is used when:

UnsafeProjection factory object is requested for a UnsafeProjection
ExpressionEncoder is requested to initialize the internal UnsafeProjection
FileFormat is requested to build a data reader with partition column values appended
OrcFileFormat is requested to build a data reader with partition column values appended
ParquetFileFormat is requested to build a data reader with partition column values appended
GroupedIterator is requested for keyProjection
ObjectOperator is requested to serializeObjectToRow
(Spark MLlib) LibSVMFileFormat is requested to buildReader
(Spark Structured Streaming) StateStoreRestoreExec, StateStoreSaveExec and StreamingDeduplicateExec are requested to execute

`canonicalize` Method



canonicalize(in: Seq[Expression]): Seq[Expression]

1

2

3

4

5

canonicalize(in: Seq[Expression]): Seq[Expression]

canonicalize removes unnecessary Alias expressions.

Internally, canonicalize uses ExpressionCanonicalizer rule executor (that in turn uses just one CleanExpressions expression rule).

Generating JVM Bytecode For UnsafeProjection For Given Expressions (With Optional Subexpression Elimination) — `create` Method



create(
  expressions: Seq[Expression],
  subexpressionEliminationEnabled: Boolean): UnsafeProjection
create(references: Seq[Expression]): UnsafeProjection (1)

1

2

3

4

5

6

7

8

create(

expressions: Seq[Expression],

subexpressionEliminationEnabled: Boolean): UnsafeProjection

create(references: Seq[Expression]): UnsafeProjection (1)

Calls the former create with subexpressionEliminationEnabled flag off

create first creates a CodegenContext and an Java source code for the input expressions.

create creates a code body with public java.lang.Object generate(Object[] references) method that creates a SpecificUnsafeProjection.



public java.lang.Object generate(Object[] references) {
  return new SpecificUnsafeProjection(references);
}

class SpecificUnsafeProjection extends UnsafeProjection {
  ...
}

1

2

3

4

5

6

7

8

9

10

11

public java.lang.Object generate(Object[] references) {

return new SpecificUnsafeProjection(references);

}

class SpecificUnsafeProjection extends UnsafeProjection {

...

}

create creates a CodeAndComment with the code body and comment placeholders.

You should see the following DEBUG message in the logs:



DEBUG GenerateUnsafeProjection: code for [expressions]:
[code]

1

2

3

4

5

6

DEBUG GenerateUnsafeProjection: code for [expressions]:

[code]

Tip

Enable DEBUG logging level for org.apache.spark.sql.catalyst.expressions.codegen.CodeGenerator logger to see the message above.



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

1

2

3

4

5

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

See CodeGenerator.

create requests CodeGenerator to compile the Java source code to JVM bytecode (using Janino).

create requests CodegenContext for references and requests the compiled class to create a SpecificUnsafeProjection for the input references that in the end is the final UnsafeProjection.

Note	(Single-argument) `create` is part of CodeGenerator Contract.

Creating ExprCode for Expressions (With Optional Subexpression Elimination) — `createCode` Method



createCode(
  ctx: CodegenContext,
  expressions: Seq[Expression],
  useSubexprElimination: Boolean = false): ExprCode

1

2

3

4

5

6

7

8

createCode(

ctx: CodegenContext,

expressions: Seq[Expression],

useSubexprElimination: Boolean = false): ExprCode

createCode requests the input CodegenContext to generate a Java source code for code-generated evaluation of every expression in the input expressions.

createCode…FIXME



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

import org.apache.spark.sql.catalyst.expressions.codegen.GenerateUnsafeProjection
val eval = GenerateUnsafeProjection.createCode(ctx, expressions, useSubexprElimination = true)

scala> println(eval.code)

        mutableStateArray1[0].reset();

        mutableStateArray2[0].write(0, ((UTF8String) references[0] /* literal */));


            mutableStateArray2[0].write(1, ((UTF8String) references[1] /* literal */));
        mutableStateArray[0].setTotalSize(mutableStateArray1[0].totalSize());


scala> println(eval.value)
mutableStateArray[0]

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

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

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

val eval = GenerateUnsafeProjection.createCode(ctx, expressions, useSubexprElimination = true)

scala> println(eval.code)

mutableStateArray1[0].reset();

mutableStateArray2[0].write(0, ((UTF8String) references[0] /* literal */));

mutableStateArray2[0].write(1, ((UTF8String) references[1] /* literal */));

mutableStateArray[0].setTotalSize(mutableStateArray1[0].totalSize());

scala> println(eval.value)

mutableStateArray[0]

Note

createCode is used when:

CreateNamedStructUnsafe is requested to generate a Java source code
GenerateUnsafeProjection is requested to create a UnsafeProjection
CodegenSupport is requested to prepareRowVar (to generate a Java source code to consume generated columns or row from a physical operator)
HashAggregateExec is requested to doProduceWithKeys and doConsumeWithKeys
BroadcastHashJoinExec is requested to genStreamSideJoinKey (when generating the Java source code for joins)

UnsafeProjection — Generic Function to Project InternalRows to UnsafeRows

2012-12-27admin阅读(1488)

UnsafeProjection — Generic Function to Encode InternalRows to UnsafeRows

UnsafeProjection is a Projection function that encodes InternalRows as UnsafeRows.



UnsafeProjection: InternalRow =[apply]=> UnsafeRow

1

2

3

4

5

UnsafeProjection: InternalRow =[apply]=> UnsafeRow

Note	Spark SQL uses `UnsafeProjection` factory object to create concrete adhoc `UnsafeProjection` instances. The base `UnsafeProjection` has no concrete named implementations and create factory methods delegate all calls to GenerateUnsafeProjection.generate in the end.

Creating UnsafeProjection — `create` Factory Method



create(schema: StructType): UnsafeProjection      (1)
create(fields: Array[DataType]): UnsafeProjection (2)
create(expr: Expression): UnsafeProjection        (3)
create(exprs: Seq[Expression], inputSchema: Seq[Attribute]): UnsafeProjection (4)
create(exprs: Seq[Expression]): UnsafeProjection  (5)
create(
  exprs: Seq[Expression],
  inputSchema: Seq[Attribute],
  subexpressionEliminationEnabled: Boolean): UnsafeProjection

1

2

3

4

5

6

7

8

9

10

11

12

13

create(schema: StructType): UnsafeProjection (1)

create(fields: Array[DataType]): UnsafeProjection (2)

create(expr: Expression): UnsafeProjection (3)

create(exprs: Seq[Expression], inputSchema: Seq[Attribute]): UnsafeProjection (4)

create(exprs: Seq[Expression]): UnsafeProjection (5)

create(

exprs: Seq[Expression],

inputSchema: Seq[Attribute],

subexpressionEliminationEnabled: Boolean): UnsafeProjection

create takes the DataTypes from schema and calls the 2nd create
create creates a BoundReference per field in fields and calls the 5th create
create calls the 5th create
create calls the 5th create
The main create that does the heavy work

create transforms all CreateNamedStruct expressions to CreateNamedStructUnsafe in every BoundReference in the input exprs.

In the end, create requests GenerateUnsafeProjection to generate a UnsafeProjection.

Note	A variant of `create` takes `subexpressionEliminationEnabled` flag (that usually is subexpressionEliminationEnabled flag of `SparkPlan`).

Projection Contract — Functions to Produce InternalRow for InternalRow

2012-12-26admin阅读(1333)

Projection Contract — Functions to Produce InternalRow for InternalRow

Projection is a contract of Scala functions that produce an internal binary row for a given internal row.



Projection: InternalRow => InternalRow

1

2

3

4

5

Projection: InternalRow => InternalRow

Projection can optionally be initialized with the current partition index (which by default does nothing).



initialize(partitionIndex: Int): Unit = {}

1

2

3

4

5

initialize(partitionIndex: Int): Unit = {}

Note	`initialize` is overriden by InterpretedProjection and `InterpretedMutableProjection` projections that are used in interpreted expression evaluation.

Table 1. Projections
Projection	Description
UnsafeProjection
InterpretedProjection
`IdentityProjection`
`MutableProjection`
`InterpretedMutableProjection`	Appears not to be used anymore

Exchange Contract

2012-12-25admin阅读(1652)

Exchange Contract — Base for Unary Physical Operators that Exchange Data

Exchange is the base of unary physical operators that exchange data among multiple threads or processes.

When requested for the output schema, Exchange simply uses the child physical operator’s output schema.

Table 1. Exchanges
Exchange	Description
BroadcastExchangeExec
ShuffleExchangeExec

spark-sql 第22页

BroadcastHashJoinExec Binary Physical Operator for Broadcast Hash Join

Executing Physical Operator (Generating RDD[InternalRow]) — doExecute Method

Generating Java Source Code for Inner Join — codegenInner Internal Method

Generating Java Source Code for Left or Right Outer Join — codegenOuter Internal Method

Generating Java Source Code for Left Semi Join — codegenSemi Internal Method

Generating Java Source Code for Anti Join — codegenAnti Internal Method

codegenExistence Internal Method

genStreamSideJoinKey Internal Method

Creating BroadcastHashJoinExec Instance

BroadcastExchangeExec Unary Physical Operator for Broadcast Joins

Waiting Until Relation Has Been Broadcast — doExecuteBroadcast Method

Lazily-Once-Initialized Asynchronously-Broadcast relationFuture Internal Attribute

Broadcasting Relation (Rows) Asynchronously — doPrepare Method

Creating BroadcastExchangeExec Instance

SQLMetric — SQL Execution Metric of Physical Operator

reset Method

Posting Driver-Side Metric Updates — SQLMetrics.postDriverMetricUpdates Method

CodeGeneratorWithInterpretedFallback

createObject Method

InterpretedProjection

Initializing Nondeterministic Expressions — initialize Method

GenerateMutableProjection

Creating MutableProjection — create Internal Method

GenerateUnsafeProjection

Generating UnsafeProjection — generate Method

canonicalize Method

Generating JVM Bytecode For UnsafeProjection For Given Expressions (With Optional Subexpression Elimination) — create Method

Creating ExprCode for Expressions (With Optional Subexpression Elimination) — createCode Method

UnsafeProjection — Generic Function to Encode InternalRows to UnsafeRows

Creating UnsafeProjection — create Factory Method

Projection Contract — Functions to Produce InternalRow for InternalRow

Exchange Contract — Base for Unary Physical Operators that Exchange Data

欢迎关注：spark技术分享

关注公众号：spark技术分享

QQ咨询

回顶部

Executing Physical Operator (Generating RDD[InternalRow]) — `doExecute` Method

Generating Java Source Code for Inner Join — `codegenInner` Internal Method

Generating Java Source Code for Left or Right Outer Join — `codegenOuter` Internal Method

Generating Java Source Code for Left Semi Join — `codegenSemi` Internal Method

Generating Java Source Code for Anti Join — `codegenAnti` Internal Method

`codegenExistence` Internal Method

`genStreamSideJoinKey` Internal Method

Waiting Until Relation Has Been Broadcast — `doExecuteBroadcast` Method

Lazily-Once-Initialized Asynchronously-Broadcast `relationFuture` Internal Attribute

Broadcasting Relation (Rows) Asynchronously — `doPrepare` Method

`reset` Method

Posting Driver-Side Metric Updates — `SQLMetrics.postDriverMetricUpdates` Method

`createObject` Method

Initializing Nondeterministic Expressions — `initialize` Method

Creating MutableProjection — `create` Internal Method

Generating UnsafeProjection — `generate` Method

`canonicalize` Method

Generating JVM Bytecode For UnsafeProjection For Given Expressions (With Optional Subexpression Elimination) — `create` Method

Creating ExprCode for Expressions (With Optional Subexpression Elimination) — `createCode` Method

Creating UnsafeProjection — `create` Factory Method