UnsafeFixedWidthAggregationMap-spark技术分享

UnsafeFixedWidthAggregationMap

UnsafeFixedWidthAggregationMap is a tiny layer (extension) around Spark Core’s BytesToBytesMap to allow for UnsafeRow keys and values.

Whenever requested for performance metrics (i.e. average number of probes per key lookup and peak memory used), UnsafeFixedWidthAggregationMap simply requests the underlying BytesToBytesMap.

UnsafeFixedWidthAggregationMap is created when:

HashAggregateExec physical operator is requested to create a new UnsafeFixedWidthAggregationMap (when HashAggregateExec physical operator is requested to generate the Java source code for “produce” path in Whole-Stage Code Generation)
TungstenAggregationIterator is created (when HashAggregateExec physical operator is requested to execute in traditional / non-Whole-Stage-Code-Generation execution path)

Table 1. UnsafeFixedWidthAggregationMap’s Internal Properties (e.g. Registries, Counters and Flags)
Name	Description
`currentAggregationBuffer`	Re-used pointer (as an UnsafeRow with the number of fields to match the aggregationBufferSchema) to the current aggregation buffer Used exclusively when `UnsafeFixedWidthAggregationMap` is requested to getAggregationBufferFromUnsafeRow.
`emptyAggregationBuffer`	Empty aggregation buffer (encoded in UnsafeRow format)
`groupingKeyProjection`	UnsafeProjection for the groupingKeySchema (to encode grouping keys as UnsafeRows)
`map`	Spark Core’s `BytesToBytesMap` with the taskMemoryManager, initialCapacity, pageSizeBytes and performance metrics enabled

`supportsAggregationBufferSchema` Static Method



boolean supportsAggregationBufferSchema(StructType schema)

boolean supportsAggregationBufferSchema(StructType schema)

supportsAggregationBufferSchema is a predicate that is enabled (true) unless there is a field (in the fields of the input schema) whose data type is not mutable.

Note

The mutable data types: BooleanType, ByteType, DateType, DecimalType, DoubleType, FloatType, IntegerType, LongType, NullType, ShortType and TimestampType.

Examples (possibly all) of data types that are not mutable: ArrayType, BinaryType, StringType, CalendarIntervalType, MapType, ObjectType and StructType.



import org.apache.spark.sql.execution.UnsafeFixedWidthAggregationMap

import org.apache.spark.sql.types._
val schemaWithImmutableField = StructType(StructField("string", StringType) :: Nil)
assert(UnsafeFixedWidthAggregationMap.supportsAggregationBufferSchema(schemaWithImmutableField) == false)

val schemaWithMutableFields = StructType(
  StructField("int", IntegerType) :: StructField("bool", BooleanType) :: Nil)
assert(UnsafeFixedWidthAggregationMap.supportsAggregationBufferSchema(schemaWithMutableFields))

import org.apache.spark.sql.execution.UnsafeFixedWidthAggregationMap

import org.apache.spark.sql.types._

val schemaWithImmutableField = StructType(StructField("string", StringType) :: Nil)

assert(UnsafeFixedWidthAggregationMap.supportsAggregationBufferSchema(schemaWithImmutableField) == false)

val schemaWithMutableFields = StructType(

StructField("int", IntegerType) :: StructField("bool", BooleanType) :: Nil)

assert(UnsafeFixedWidthAggregationMap.supportsAggregationBufferSchema(schemaWithMutableFields))

Note	`supportsAggregationBufferSchema` is used exclusively when `HashAggregateExec` is requested to supportsAggregate.

Creating UnsafeFixedWidthAggregationMap Instance

UnsafeFixedWidthAggregationMap takes the following when created:

Empty aggregation buffer (as an InternalRow)
Aggregation buffer schema
Grouping key schema
Spark Core’s TaskMemoryManager
Initial capacity
Page size (in bytes)

UnsafeFixedWidthAggregationMap initializes the internal registries and counters.

`getAggregationBufferFromUnsafeRow` Method



UnsafeRow getAggregationBufferFromUnsafeRow(UnsafeRow key) (1)
UnsafeRow getAggregationBufferFromUnsafeRow(UnsafeRow key, int hash)

UnsafeRow getAggregationBufferFromUnsafeRow(UnsafeRow key) (1)

UnsafeRow getAggregationBufferFromUnsafeRow(UnsafeRow key, int hash)

Uses the hash code of the key

getAggregationBufferFromUnsafeRow requests the BytesToBytesMap to lookup the input key (to get a BytesToBytesMap.Location).

getAggregationBufferFromUnsafeRow…FIXME

Note	`getAggregationBufferFromUnsafeRow` is used when: `TungstenAggregationIterator` is requested to processInputs (exclusively when `TungstenAggregationIterator` is created) (for testing only) `UnsafeFixedWidthAggregationMap` is requested to getAggregationBuffer

`getAggregationBuffer` Method



UnsafeRow getAggregationBuffer(InternalRow groupingKey)

UnsafeRow getAggregationBuffer(InternalRow groupingKey)

getAggregationBuffer…FIXME

Note	`getAggregationBuffer` seems to be used exclusively for testing.

Getting KVIterator — `iterator` Method



KVIterator<UnsafeRow, UnsafeRow> iterator()

KVIterator<UnsafeRow, UnsafeRow> iterator()

iterator…FIXME

Note	`iterator` is used when: `HashAggregateExec` physical operator is requested to finishAggregate `TungstenAggregationIterator` is created (and pre-loads the first key-value pair from the map)

`getPeakMemoryUsedBytes` Method



long getPeakMemoryUsedBytes()

long getPeakMemoryUsedBytes()

getPeakMemoryUsedBytes…FIXME

Note	`getPeakMemoryUsedBytes` is used when: `HashAggregateExec` physical operator is requested to finishAggregate `TungstenAggregationIterator` is used in TaskCompletionListener

`getAverageProbesPerLookup` Method



double getAverageProbesPerLookup()

double getAverageProbesPerLookup()

getAverageProbesPerLookup…FIXME

Note	`getAverageProbesPerLookup` is used when: `HashAggregateExec` physical operator is requested to finishAggregate `TungstenAggregationIterator` is used in TaskCompletionListener

`free` Method



void free()

void free()

free…FIXME

Note	`free` is used when: `HashAggregateExec` physical operator is requested to finishAggregate `TungstenAggregationIterator` is requested to processInputs (when `TungstenAggregationIterator` is created), get the next UnsafeRow, outputForEmptyGroupingKeyWithoutInput and is created

`destructAndCreateExternalSorter` Method



UnsafeKVExternalSorter destructAndCreateExternalSorter() throws IOException

UnsafeKVExternalSorter destructAndCreateExternalSorter() throws IOException

destructAndCreateExternalSorter…FIXME

Note	`destructAndCreateExternalSorter` is used when: `HashAggregateExec` physical operator is requested to finishAggregate `TungstenAggregationIterator` is requested to processInputs (when `TungstenAggregationIterator` is created)

UnsafeFixedWidthAggregationMap