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

SizeEstimator

SizeEstimator is…FIXME

`estimate` Method



estimate(obj: AnyRef): Long (1)
// Internal
estimate(obj: AnyRef, visited: IdentityHashMap[AnyRef, AnyRef]): Long

1

2

3

4

5

6

7

estimate(obj: AnyRef): Long (1)

// Internal

estimate(obj: AnyRef, visited: IdentityHashMap[AnyRef, AnyRef]): Long

Uses estimate with an empty IdentityHashMap

estimate…FIXME

Note	`estimate` is used when: `SizeEstimator` is requested to sampleArray FIXME

`sampleArray` Internal Method



sampleArray(
  array: AnyRef,
  state: SearchState,
  rand: Random,
  drawn: OpenHashSet[Int],
  length: Int): Long

1

2

3

4

5

6

7

8

9

10

sampleArray(

array: AnyRef,

state: SearchState,

rand: Random,

drawn: OpenHashSet[Int],

length: Int): Long

sampleArray…FIXME

Note	`sampleArray` is used when…FIXME

`visitSingleObject` Internal Method



visitSingleObject(obj: AnyRef, state: SearchState): Unit

1

2

3

4

5

visitSingleObject(obj: AnyRef, state: SearchState): Unit

visitSingleObject…FIXME

Note	`visitSingleObject` is used exclusively when `SizeEstimator` is requested to estimate.

KnownSizeEstimation

KnownSizeEstimation is the contract that allows a class to give SizeEstimator a more accurate size estimation.

KnownSizeEstimation defines the single estimatedSize method.



package org.apache.spark.util

trait KnownSizeEstimation {
  def estimatedSize: Long
}

1

2

3

4

5

6

7

8

9

package org.apache.spark.util

trait KnownSizeEstimation {

def estimatedSize: Long

}

estimatedSize is used when:

SizeEstimator is requested to visitSingleObject
BroadcastExchangeExec is requested for relationFuture
BroadcastHashJoinExec is requested to execute
ShuffledHashJoinExec is requested to buildHashedRelation

Note	`KnownSizeEstimation` is a `private[spark]` contract.

Note	HashedRelation is the only `KnownSizeEstimation` available.

UnsafeHashedRelation

UnsafeHashedRelation is…FIXME

`get` Method



get(key: InternalRow): Iterator[InternalRow]

1

2

3

4

5

get(key: InternalRow): Iterator[InternalRow]

Note	`get` is part of HashedRelation Contract to give the internal rows for the given key or `null`.

get…FIXME

Getting Value Row for Given Key — `getValue` Method



getValue(key: InternalRow): InternalRow

1

2

3

4

5

getValue(key: InternalRow): InternalRow

Note	`getValue` is part of HashedRelation Contract to give the value internal row for a given key.

getValue…FIXME

Creating UnsafeHashedRelation Instance — `apply` Factory Method



apply(
  input: Iterator[InternalRow],
  key: Seq[Expression],
  sizeEstimate: Int,
  taskMemoryManager: TaskMemoryManager): HashedRelation

1

2

3

4

5

6

7

8

9

apply(

input: Iterator[InternalRow],

key: Seq[Expression],

sizeEstimate: Int,

taskMemoryManager: TaskMemoryManager): HashedRelation

apply…FIXME

Note	`apply` is used when…FIXME

LongHashedRelation

LongHashedRelation is a HashedRelation that is used when HashedRelation is requested for a concrete HashedRelation instance when the single key is of type long.

LongHashedRelation is also a Java Externalizable, i.e. when persisted, only the identity is written in the serialization stream and it is the responsibility of the class to save and restore the contents of its instances.

LongHashedRelation is created when:

HashedRelation is requested for a concrete HashedRelation (and apply factory method is used)
LongHashedRelation is requested for a read-only copy (when BroadcastHashJoinExec is requested to execute)

`writeExternal` Method



writeExternal(out: ObjectOutput): Unit

1

2

3

4

5

writeExternal(out: ObjectOutput): Unit

Note	`writeExternal` is part of Java’s Externalizable Contract to…FIXME.

writeExternal…FIXME

Note	`writeExternal` is used when…FIXME

`readExternal` Method



readExternal(in: ObjectInput): Unit

1

2

3

4

5

readExternal(in: ObjectInput): Unit

Note	`readExternal` is part of Java’s Externalizable Contract to…FIXME.

readExternal…FIXME

Note	`readExternal` is used when…FIXME

Creating LongHashedRelation Instance

LongHashedRelation takes the following when created:

Number of fields
LongToUnsafeRowMap

LongHashedRelation initializes the internal registries and counters.

Creating Read-Only Copy of LongHashedRelation — `asReadOnlyCopy` Method



asReadOnlyCopy(): LongHashedRelation

1

2

3

4

5

asReadOnlyCopy(): LongHashedRelation

Note	`asReadOnlyCopy` is part of HashedRelation Contract to…FIXME.

asReadOnlyCopy…FIXME

Getting Value Row for Given Key — `getValue` Method



getValue(key: InternalRow): InternalRow

1

2

3

4

5

getValue(key: InternalRow): InternalRow

Note	`getValue` is part of HashedRelation Contract to give the value internal row for a given key.

getValue checks if the input key is null at 0 position and if so gives null. Otherwise, getValue takes the long value at position 0 and gets the value.

Creating LongHashedRelation Instance — `apply` Factory Method



apply(
  input: Iterator[InternalRow],
  key: Seq[Expression],
  sizeEstimate: Int,
  taskMemoryManager: TaskMemoryManager): LongHashedRelation

1

2

3

4

5

6

7

8

9

apply(

input: Iterator[InternalRow],

key: Seq[Expression],

sizeEstimate: Int,

taskMemoryManager: TaskMemoryManager): LongHashedRelation

apply…FIXME

Note	`apply` is used exclusively when `HashedRelation` is requested for a concrete HashedRelation.

HashedRelation

2013-06-27admin阅读(1864)

HashedRelation

HashedRelation is the contract for “relations” with values hashed by some key.

HashedRelation is a KnownSizeEstimation.



package org.apache.spark.sql.execution.joins

trait HashedRelation extends KnownSizeEstimation {
  // only required methods that have no implementation
  // the others follow
  def asReadOnlyCopy(): HashedRelation
  def close(): Unit
  def get(key: InternalRow): Iterator[InternalRow]
  def getAverageProbesPerLookup: Double
  def getValue(key: InternalRow): InternalRow
  def keyIsUnique: Boolean
}

1

2

3

4

5

6

7

8

9

10

11

12

13

14

15

16

package org.apache.spark.sql.execution.joins

trait HashedRelation extends KnownSizeEstimation {

// only required methods that have no implementation

// the others follow

def asReadOnlyCopy(): HashedRelation

def close(): Unit

def get(key: InternalRow): Iterator[InternalRow]

def getAverageProbesPerLookup: Double

def getValue(key: InternalRow): InternalRow

def keyIsUnique: Boolean

}

Note	`HashedRelation` is a `private[execution]` contract.

Method Description

asReadOnlyCopy

Gives a read-only copy of this HashedRelation to be safely used in a separate thread.

Used exclusively when BroadcastHashJoinExec is requested to execute (and transform every partitions of streamedPlan physical operator using the broadcast variable of buildPlan physical operator).

get

Gives internal rows for the given key or null

Used when HashJoin is requested to innerJoin, outerJoin, semiJoin, existenceJoin and antiJoin.

getValue

Gives the value internal row for a given key

Note	`HashedRelation` has two variants of `getValue`, i.e. one that accepts an `InternalRow` and another a `Long`. `getValue` with an `InternalRow` does not seem to be used at all.

getAverageProbesPerLookup

Used when…FIXME

`getValue` Method



getValue(key: Long): InternalRow

1

2

3

4

5

getValue(key: Long): InternalRow

Note	This is `getValue` that takes a long key. There is the more generic getValue that takes an internal row instead.

getValue simply reports an UnsupportedOperationException (and expects concrete HashedRelations to provide a more meaningful implementation).

Note	`getValue` is used exclusively when `LongHashedRelation` is requested to get the value for a given key.

Creating Concrete HashedRelation Instance (for Build Side of Hash-based Join) — `apply` Factory Method



apply(
  input: Iterator[InternalRow],
  key: Seq[Expression],
  sizeEstimate: Int = 64,
  taskMemoryManager: TaskMemoryManager = null): HashedRelation

1

2

3

4

5

6

7

8

9

apply(

input: Iterator[InternalRow],

key: Seq[Expression],

sizeEstimate: Int = 64,

taskMemoryManager: TaskMemoryManager = null): HashedRelation

apply creates a LongHashedRelation when the input key collection has a single expression of type long or UnsafeHashedRelation otherwise.

Note	The input `key` expressions are: Build join keys of `ShuffledHashJoinExec` physical operator Canonicalized build-side join keys of `HashedRelationBroadcastMode` (of BroadcastHashJoinExec physical operator)

Note	`apply` is used when: `ShuffledHashJoinExec` is requested to build a HashedRelation for given internal rows `HashedRelationBroadcastMode` is requested to transform

HashJoin — Contract for Hash-based Join Physical Operators

2013-06-26admin阅读(1857)

HashJoin — Contract for Hash-based Join Physical Operators

HashJoin is the contract for hash-based join physical operators (e.g. BroadcastHashJoinExec and ShuffledHashJoinExec).



package org.apache.spark.sql.execution.joins

trait HashJoin {
  // only required methods that have no implementation
  // the others follow
  val leftKeys: Seq[Expression]
  val rightKeys: Seq[Expression]
  val joinType: JoinType
  val buildSide: BuildSide
  val condition: Option[Expression]
  val left: SparkPlan
  val right: SparkPlan
}

1

2

3

4

5

6

7

8

9

10

11

12

13

14

15

16

17

package org.apache.spark.sql.execution.joins

trait HashJoin {

// only required methods that have no implementation

// the others follow

val leftKeys: Seq[Expression]

val rightKeys: Seq[Expression]

val joinType: JoinType

val buildSide: BuildSide

val condition: Option[Expression]

val left: SparkPlan

val right: SparkPlan

}

Table 1. HashJoin Contract
Method	Description
`buildSide`	Left or right build side Used when: `HashJoin` is requested for buildPlan, streamedPlan, buildKeys and streamedKeys `BroadcastHashJoinExec` physical operator is requested for requiredChildDistribution, to codegenInner and codegenOuter
`joinType`	JoinType

Table 2. HashJoin’s Internal Properties (e.g. Registries, Counters and Flags)
Name	Description
`boundCondition`
`buildKeys`	Build join keys (as Catalyst expressions)
`buildPlan`
`streamedKeys`	Streamed join keys (as Catalyst expressions)
`streamedPlan`

`join` Method



join(
  streamedIter: Iterator[InternalRow],
  hashed: HashedRelation,
  numOutputRows: SQLMetric,
  avgHashProbe: SQLMetric): Iterator[InternalRow]

1

2

3

4

5

6

7

8

9

join(

streamedIter: Iterator[InternalRow],

hashed: HashedRelation,

numOutputRows: SQLMetric,

avgHashProbe: SQLMetric): Iterator[InternalRow]

join branches off per joinType to create a join iterator of internal rows (i.e. Iterator[InternalRow]) for the input streamedIter and hashed:

innerJoin for a InnerLike join
outerJoin for a LeftOuter or a RightOuter join
semiJoin for a LeftSemi join
antiJoin for a LeftAnti join
existenceJoin for a ExistenceJoin join

join requests TaskContext to add a TaskCompletionListener to update the input avg hash probe SQL metric. The TaskCompletionListener is executed on a task completion (regardless of the task status: success, failure, or cancellation) and uses getAverageProbesPerLookup from the input hashed to set the input avg hash probe.

join createResultProjection.

In the end, for every row in the join iterator of internal rows join increments the input numOutputRows SQL metric and applies the result projection.

join reports a IllegalArgumentException when the joinType is incorrect.



[x] JoinType is not supported

1

2

3

4

5

[x] JoinType is not supported

Note	`join` is used when BroadcastHashJoinExec and ShuffledHashJoinExec are executed.

`innerJoin` Internal Method



innerJoin(
  streamIter: Iterator[InternalRow],
  hashedRelation: HashedRelation): Iterator[InternalRow]

1

2

3

4

5

6

7

innerJoin(

streamIter: Iterator[InternalRow],

hashedRelation: HashedRelation): Iterator[InternalRow]

innerJoin…FIXME

Note	`innerJoin` is used when…FIXME

`outerJoin` Internal Method



outerJoin(
  streamedIter: Iterator[InternalRow],
  hashedRelation: HashedRelation): Iterator[InternalRow]

1

2

3

4

5

6

7

outerJoin(

streamedIter: Iterator[InternalRow],

hashedRelation: HashedRelation): Iterator[InternalRow]

outerJoin…FIXME

Note	`outerJoin` is used when…FIXME

`semiJoin` Internal Method



semiJoin(
  streamIter: Iterator[InternalRow],
  hashedRelation: HashedRelation): Iterator[InternalRow]

1

2

3

4

5

6

7

semiJoin(

streamIter: Iterator[InternalRow],

hashedRelation: HashedRelation): Iterator[InternalRow]

semiJoin…FIXME

Note	`semiJoin` is used when…FIXME

`antiJoin` Internal Method



antiJoin(
  streamIter: Iterator[InternalRow],
  hashedRelation: HashedRelation): Iterator[InternalRow]

1

2

3

4

5

6

7

antiJoin(

streamIter: Iterator[InternalRow],

hashedRelation: HashedRelation): Iterator[InternalRow]

antiJoin…FIXME

Note	`antiJoin` is used when…FIXME

`existenceJoin` Internal Method



existenceJoin(
  streamIter: Iterator[InternalRow],
  hashedRelation: HashedRelation): Iterator[InternalRow]

1

2

3

4

5

6

7

existenceJoin(

streamIter: Iterator[InternalRow],

hashedRelation: HashedRelation): Iterator[InternalRow]

existenceJoin…FIXME

Note	`existenceJoin` is used when…FIXME

`createResultProjection` Method



createResultProjection(): (InternalRow) => InternalRow

1

2

3

4

5

createResultProjection(): (InternalRow) => InternalRow

createResultProjection…FIXME

Note	`createResultProjection` is used when…FIXME

PhysicalOperation — Scala Extractor for Destructuring Logical Query Plans

2013-06-25admin阅读(4233)

PhysicalOperation — Scala Extractor for Destructuring Logical Query Plans

PhysicalOperation is a Scala extractor to destructure a logical query plan into a tuple with the following elements:

Named expressions (aka projects)
Expressions (aka filters)
Logical operator (aka leaf node)


ReturnType

(Seq[NamedExpression], Seq[Expression], LogicalPlan)

1

2

3

4

5

6

ReturnType

(Seq[NamedExpression], Seq[Expression], LogicalPlan)

The following idiom is often used in Strategy implementations (e.g. HiveTableScans, InMemoryScans, DataSourceStrategy, FileSourceStrategy):



def apply(plan: LogicalPlan): Seq[SparkPlan] = plan match {
  case PhysicalOperation(projections, predicates, plan) =>
    // do something
  case _ => Nil
}

1

2

3

4

5

6

7

8

9

def apply(plan: LogicalPlan): Seq[SparkPlan] = plan match {

case PhysicalOperation(projections, predicates, plan) =>

// do something

case _ => Nil

}

Whenever used to pattern match to a LogicalPlan, PhysicalOperation‘s unapply is called.

`unapply` Method



type ReturnType = (Seq[NamedExpression], Seq[Expression], LogicalPlan)

unapply(plan: LogicalPlan): Option[ReturnType]

1

2

3

4

5

6

7

type ReturnType = (Seq[NamedExpression], Seq[Expression], LogicalPlan)

unapply(plan: LogicalPlan): Option[ReturnType]

unapply…FIXME

Note	`unapply` is almost collectProjectsAndFilters method itself (with some manipulations of the return value).

Note	`unapply` is used when…FIXME

PhysicalAggregation — Scala Extractor for Destructuring Aggregate Logical Operators

2013-06-24admin阅读(1866)

PhysicalAggregation — Scala Extractor for Destructuring Aggregate Logical Operators

PhysicalAggregation is a Scala extractor to destructure an Aggregate logical operator into a four-element tuple with the following elements:


ReturnType

(Seq[NamedExpression], Seq[AggregateExpression], Seq[NamedExpression], LogicalPlan)

1

2

3

4

5

6

ReturnType

(Seq[NamedExpression], Seq[AggregateExpression], Seq[NamedExpression], LogicalPlan)

Tip	See the document about Scala extractor objects.

Destructuring Aggregate Logical Operator — `unapply` Method



type ReturnType =
  (Seq[NamedExpression], Seq[AggregateExpression], Seq[NamedExpression], LogicalPlan)

unapply(a: Any): Option[ReturnType]

1

2

3

4

5

6

7

8

type ReturnType =

(Seq[NamedExpression], Seq[AggregateExpression], Seq[NamedExpression], LogicalPlan)

unapply(a: Any): Option[ReturnType]

unapply destructures the input a Aggregate logical operator into a four-element ReturnType.

Note	`unapply` is used when…FIXME

TypeCoercionRule — Contract For Type Coercion Rules

2013-06-23admin阅读(1966)

TypeCoercionRule Contract — Type Coercion Rules

TypeCoercionRule is the contract of logical rules to coerce and propagate types in logical plans.



package org.apache.spark.sql.catalyst.analysis

trait TypeCoercionRule extends Rule[LogicalPlan] {
  // only required methods that have no implementation
  // the others follow
  def coerceTypes(plan: LogicalPlan): LogicalPlan
}

1

2

3

4

5

6

7

8

9

10

11

package org.apache.spark.sql.catalyst.analysis

trait TypeCoercionRule extends Rule[LogicalPlan] {

// only required methods that have no implementation

// the others follow

def coerceTypes(plan: LogicalPlan): LogicalPlan

}

Table 1. (Subset of) TypeCoercionRule Contract
Method	Description
`coerceTypes`	Coerce types in a logical plan Used exclusively when `TypeCoercionRule` is executed

TypeCoercionRule is simply a Catalyst rule for transforming logical plans, i.e. Rule[LogicalPlan].

Table 2. TypeCoercionRules
TypeCoercionRule	Description
CaseWhenCoercion
ConcatCoercion
DecimalPrecision
Division
EltCoercion
FunctionArgumentConversion
IfCoercion
ImplicitTypeCasts
InConversion
PromoteStrings
StackCoercion
WindowFrameCoercion

Executing Rule — `apply` Method



apply(plan: LogicalPlan): LogicalPlan

1

2

3

4

5

apply(plan: LogicalPlan): LogicalPlan

Note	`apply` is part of the Rule Contract to execute (apply) a rule on a TreeNode (e.g. LogicalPlan).

apply coerceTypes in the input LogicalPlan and returns the following:

The input LogicalPlan itself if matches the coerced plan
The plan after propagateTypes on the coerced plan

`propagateTypes` Internal Method



propagateTypes(plan: LogicalPlan): LogicalPlan

1

2

3

4

5

propagateTypes(plan: LogicalPlan): LogicalPlan

propagateTypes…FIXME

Note	`propagateTypes` is used exclusively when `TypeCoercionRule` is executed.

TypeCoercion Object

2013-06-22admin阅读(1490)

TypeCoercion Object

TypeCoercion is a Scala object that defines the type coercion rules for Spark Analyzer.

Defining Type Coercion Rules (For Spark Analyzer) — `typeCoercionRules` Method



typeCoercionRules(conf: SQLConf): List[Rule[LogicalPlan]]

1

2

3

4

5

typeCoercionRules(conf: SQLConf): List[Rule[LogicalPlan]]

typeCoercionRules is a collection of Catalyst rules to transform logical plans (in the order of execution):

InConversion
WidenSetOperationTypes
PromoteStrings
DecimalPrecision
BooleanEquality
FunctionArgumentConversion
ConcatCoercion
EltCoercion
CaseWhenCoercion
IfCoercion
StackCoercion
Division
ImplicitTypeCasts
DateTimeOperations
WindowFrameCoercion

Note	`typeCoercionRules` is used exclusively when `Analyzer` is requested for batches.

spark-sql 第4页

SizeEstimator

estimate Method

sampleArray Internal Method

visitSingleObject Internal Method

KnownSizeEstimation

UnsafeHashedRelation

get Method

Getting Value Row for Given Key — getValue Method

Creating UnsafeHashedRelation Instance — apply Factory Method

LongHashedRelation

writeExternal Method

readExternal Method

Creating LongHashedRelation Instance

Creating Read-Only Copy of LongHashedRelation — asReadOnlyCopy Method

Getting Value Row for Given Key — getValue Method

Creating LongHashedRelation Instance — apply Factory Method

HashedRelation

getValue Method

Creating Concrete HashedRelation Instance (for Build Side of Hash-based Join) — apply Factory Method

HashJoin — Contract for Hash-based Join Physical Operators

join Method

innerJoin Internal Method

outerJoin Internal Method

semiJoin Internal Method

antiJoin Internal Method

existenceJoin Internal Method

createResultProjection Method

PhysicalOperation — Scala Extractor for Destructuring Logical Query Plans

unapply Method

PhysicalAggregation — Scala Extractor for Destructuring Aggregate Logical Operators

Destructuring Aggregate Logical Operator — unapply Method

TypeCoercionRule Contract — Type Coercion Rules

Executing Rule — apply Method

propagateTypes Internal Method

TypeCoercion Object

Defining Type Coercion Rules (For Spark Analyzer) — typeCoercionRules Method

欢迎关注：spark技术分享

关注公众号：spark技术分享

QQ咨询

回顶部

`estimate` Method

`sampleArray` Internal Method

`visitSingleObject` Internal Method

`get` Method

Getting Value Row for Given Key — `getValue` Method

Creating UnsafeHashedRelation Instance — `apply` Factory Method

`writeExternal` Method

`readExternal` Method

Creating Read-Only Copy of LongHashedRelation — `asReadOnlyCopy` Method

Getting Value Row for Given Key — `getValue` Method

Creating LongHashedRelation Instance — `apply` Factory Method

`getValue` Method

Creating Concrete HashedRelation Instance (for Build Side of Hash-based Join) — `apply` Factory Method

`join` Method

`innerJoin` Internal Method

`outerJoin` Internal Method

`semiJoin` Internal Method

`antiJoin` Internal Method

`existenceJoin` Internal Method

`createResultProjection` Method

`unapply` Method

Destructuring Aggregate Logical Operator — `unapply` Method

Executing Rule — `apply` Method

`propagateTypes` Internal Method

Defining Type Coercion Rules (For Spark Analyzer) — `typeCoercionRules` Method