org.apache.flink.runtime.state

Class KeyGroupPartitioner<T>

java.lang.Object

org.apache.flink.runtime.state.KeyGroupPartitioner<T>

Type Parameters:

T - type of the partitioned elements.

Direct Known Subclasses:

CopyOnWriteStateTableSnapshot.StateTableKeyGroupPartitioner

public class KeyGroupPartitioner<T>
extends Object

Class that contains the base algorithm for partitioning data into key-groups. This algorithm currently works with two array (input, output) for optimal algorithmic complexity. Notice that this could also be implemented over a single array, using some cuckoo-hashing-style element replacement. This would have worse algorithmic complexity but better space efficiency. We currently prefer the trade-off in favor of better algorithmic complexity.

Nested Class Summary

Nested Classes

Modifier and Type	Class and Description
static interface	KeyGroupPartitioner.ElementReaderFunction<T> This functional interface defines how one element is read from a DataInputView.
static interface	KeyGroupPartitioner.ElementWriterFunction<T> This functional interface defines how one element is written to a DataOutputView.
static interface	KeyGroupPartitioner.KeyGroupElementsConsumer<T> Functional interface to consume elements from a key group.

Field Summary

Fields

Modifier and Type	Field and Description
protected StateSnapshot.StateKeyGroupWriter	computedResult Cached result.
protected int[]	counterHistogram This bookkeeping array is used to count the elements in each key-group.
protected int[]	elementKeyGroups This is a helper array that caches the key-group for each element, so we do not have to compute them twice.
protected KeyGroupPartitioner.ElementWriterFunction<T>	elementWriterFunction Function to write an element to a DataOutputView.
protected int	firstKeyGroup Cached value of keyGroupRange#firstKeyGroup.
protected KeyExtractorFunction<T>	keyExtractorFunction Function to extract the key from a given element.
protected KeyGroupRange	keyGroupRange The key-group range for the input data, covered in this partitioning.
protected int	numberOfElements Total number of input elements.
protected T[]	partitioningDestination The output array for the partitioning.
protected T[]	partitioningSource The input data for the partitioning.
protected int	totalKeyGroups The total number of key-groups in the job.

Constructor Summary

Constructors

Constructor and Description
KeyGroupPartitioner(T[] partitioningSource, int numberOfElements, T[] partitioningDestination, KeyGroupRange keyGroupRange, int totalKeyGroups, KeyExtractorFunction<T> keyExtractorFunction, KeyGroupPartitioner.ElementWriterFunction<T> elementWriterFunction) Creates a new KeyGroupPartitioner.

Method Summary

Modifier and Type	Method and Description
static <T> StateSnapshotKeyGroupReader	createKeyGroupPartitionReader(KeyGroupPartitioner.ElementReaderFunction<T> readerFunction, KeyGroupPartitioner.KeyGroupElementsConsumer<T> elementConsumer)
StateSnapshot.StateKeyGroupWriter	partitionByKeyGroup() Partitions the data into key-groups and returns the result via PartitioningResult.
protected void	reportAllElementKeyGroups() This method iterates over the input data and reports the key-group for each element.
protected void	reportKeyGroupOfElementAtIndex(int index, int keyGroup) This method reports in the bookkeeping data that the element at the given index belongs to the given key-group.

Methods inherited from class java.lang.Object

clone, equals, finalize, getClass, hashCode, notify, notifyAll, toString, wait, wait, wait

Field Detail

partitioningSource

@Nonnull
protected final T[] partitioningSource

The input data for the partitioning. All elements to consider must be densely in the index interval [0, numberOfElements[, without null values.

partitioningDestination

@Nonnull
protected final T[] partitioningDestination

The output array for the partitioning. The size must be numberOfElements (or bigger).

numberOfElements

@Nonnegative
protected final int numberOfElements

Total number of input elements.

totalKeyGroups

@Nonnegative
protected final int totalKeyGroups

The total number of key-groups in the job.

keyGroupRange

@Nonnull
protected final KeyGroupRange keyGroupRange

The key-group range for the input data, covered in this partitioning.

counterHistogram

@Nonnull
protected final int[] counterHistogram

This bookkeeping array is used to count the elements in each key-group. In a second step, it is transformed into a histogram by accumulation.

elementKeyGroups

@Nonnull
protected final int[] elementKeyGroups

This is a helper array that caches the key-group for each element, so we do not have to compute them twice.

firstKeyGroup

@Nonnegative
protected final int firstKeyGroup

Cached value of keyGroupRange#firstKeyGroup.

keyExtractorFunction

@Nonnull
protected final KeyExtractorFunction<T> keyExtractorFunction

Function to extract the key from a given element.

elementWriterFunction

@Nonnull
protected final KeyGroupPartitioner.ElementWriterFunction<T> elementWriterFunction

Function to write an element to a DataOutputView.

computedResult

@Nullable
protected StateSnapshot.StateKeyGroupWriter computedResult

Cached result.

Constructor Detail

KeyGroupPartitioner

public KeyGroupPartitioner(@Nonnull
                           T[] partitioningSource,
                           @Nonnegative
                           int numberOfElements,
                           @Nonnull
                           T[] partitioningDestination,
                           @Nonnull
                           KeyGroupRange keyGroupRange,
                           @Nonnegative
                           int totalKeyGroups,
                           @Nonnull
                           KeyExtractorFunction<T> keyExtractorFunction,
                           @Nonnull
                           KeyGroupPartitioner.ElementWriterFunction<T> elementWriterFunction)

Creates a new KeyGroupPartitioner.

Parameters:

partitioningSource - the input for the partitioning. All elements must be densely packed in the index interval [0, numberOfElements[, without null values.

numberOfElements - the number of elements to consider from the input, starting at input index 0.

partitioningDestination - the output of the partitioning. Must have capacity of at least numberOfElements.

keyGroupRange - the key-group range of the data that will be partitioned by this instance.

totalKeyGroups - the total number of key groups in the job.

keyExtractorFunction - this function extracts the partition key from an element.

Method Detail

partitionByKeyGroup

public StateSnapshot.StateKeyGroupWriter partitionByKeyGroup()

Partitions the data into key-groups and returns the result via PartitioningResult.

reportAllElementKeyGroups

protected void reportAllElementKeyGroups()

This method iterates over the input data and reports the key-group for each element.

reportKeyGroupOfElementAtIndex

protected void reportKeyGroupOfElementAtIndex(int index,
                                              int keyGroup)

This method reports in the bookkeeping data that the element at the given index belongs to the given key-group.

createKeyGroupPartitionReader

public static <T> StateSnapshotKeyGroupReader createKeyGroupPartitionReader(@Nonnull
                                                                            KeyGroupPartitioner.ElementReaderFunction<T> readerFunction,
                                                                            @Nonnull
                                                                            KeyGroupPartitioner.KeyGroupElementsConsumer<T> elementConsumer)