org.apache.flink.runtime.operators.sort

Class CombiningUnilateralSortMerger<E>

java.lang.Object

org.apache.flink.runtime.operators.sort.UnilateralSortMerger<E>

org.apache.flink.runtime.operators.sort.CombiningUnilateralSortMerger<E>

All Implemented Interfaces:

Closeable, AutoCloseable, Sorter<E>, CloseableInputProvider<E>

public class CombiningUnilateralSortMerger<E>
extends UnilateralSortMerger<E>

The CombiningUnilateralSortMerger is part of a merge-sort implementation. Conceptually, a merge sort with combining works as follows: (1) Divide the unsorted list into n sublists of about 1/n the size. (2) Sort each sublist recursively by re-applying merge sort. (3) Combine all tuples with the same key within a sublist (4) Merge the two sublists back into one sorted list. Internally, the CombiningUnilateralSortMerger logic is factored into four threads (read, sort, combine, spill) which communicate through a set of blocking queues (forming a closed loop). Memory is allocated using the MemoryManager interface. Thus the component will most likely not exceed the user-provided memory limits.

Nested Class Summary

Nested Classes

Modifier and Type	Class and Description
protected class	CombiningUnilateralSortMerger.CombiningSpillingThread

Nested classes/interfaces inherited from class org.apache.flink.runtime.operators.sort.UnilateralSortMerger

UnilateralSortMerger.ChannelWithBlockCount, UnilateralSortMerger.CircularElement<E>, UnilateralSortMerger.CircularQueues<E>, UnilateralSortMerger.ReadingThread<E>, UnilateralSortMerger.SortingThread<E>, UnilateralSortMerger.SpillingThread, UnilateralSortMerger.ThreadBase<E>

Field Summary

Fields inherited from class org.apache.flink.runtime.operators.sort.UnilateralSortMerger

closed, iterator, iteratorException, iteratorLock, MAX_NUM_WRITE_BUFFERS, memoryManager, MIN_NUM_SORT_MEM_SEGMENTS, MIN_NUM_WRITE_BUFFERS, objectReuseEnabled, sortReadMemory, writeMemory

Constructor Summary

Constructors

Constructor and Description
CombiningUnilateralSortMerger(GroupCombineFunction<E,E> combineStub, MemoryManager memoryManager, IOManager ioManager, MutableObjectIterator<E> input, AbstractInvokable parentTask, TypeSerializerFactory<E> serializerFactory, TypeComparator<E> comparator, double memoryFraction, int maxNumFileHandles, float startSpillingFraction, boolean objectReuseEnabled) Creates a new sorter that reads the data from a given reader and provides an iterator returning that data in a sorted manner.
CombiningUnilateralSortMerger(GroupCombineFunction<E,E> combineStub, MemoryManager memoryManager, IOManager ioManager, MutableObjectIterator<E> input, AbstractInvokable parentTask, TypeSerializerFactory<E> serializerFactory, TypeComparator<E> comparator, double memoryFraction, int numSortBuffers, int maxNumFileHandles, float startSpillingFraction, boolean objectReuseEnabled) Creates a new sorter that reads the data from a given reader and provides an iterator returning that data in a sorted manner.

Method Summary

Modifier and Type	Method and Description
protected UnilateralSortMerger.ThreadBase<E>	getSpillingThread(ExceptionHandler<IOException> exceptionHandler, UnilateralSortMerger.CircularQueues<E> queues, AbstractInvokable parentTask, MemoryManager memoryManager, IOManager ioManager, TypeSerializerFactory<E> serializerFactory, TypeComparator<E> comparator, List<MemorySegment> sortReadMemory, List<MemorySegment> writeMemory, int maxFileHandles)
void	setUdfConfiguration(Configuration config)

Methods inherited from class org.apache.flink.runtime.operators.sort.UnilateralSortMerger

close, endMarker, getIterator, getReadingThread, getSortingThread, setResultIterator, setResultIteratorException, spillingMarker, startThreads

Methods inherited from class java.lang.Object

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

Constructor Detail

CombiningUnilateralSortMerger

public CombiningUnilateralSortMerger(GroupCombineFunction<E,E> combineStub,
                                     MemoryManager memoryManager,
                                     IOManager ioManager,
                                     MutableObjectIterator<E> input,
                                     AbstractInvokable parentTask,
                                     TypeSerializerFactory<E> serializerFactory,
                                     TypeComparator<E> comparator,
                                     double memoryFraction,
                                     int maxNumFileHandles,
                                     float startSpillingFraction,
                                     boolean objectReuseEnabled)
                              throws IOException,
                                     MemoryAllocationException

Creates a new sorter that reads the data from a given reader and provides an iterator returning that data in a sorted manner. The memory is divided among sort buffers, write buffers and read buffers automatically.

Parameters:

combineStub - The stub used to combine values with the same key.

memoryManager - The memory manager from which to allocate the memory.

ioManager - The I/O manager, which is used to write temporary files to disk.

input - The input that is sorted by this sorter.

parentTask - The parent task, which owns all resources used by this sorter.

serializerFactory - The type serializer.

comparator - The type comparator establishing the order relation.

memoryFraction - The fraction of memory dedicated to sorting, merging and I/O.

maxNumFileHandles - The maximum number of files to be merged at once.

startSpillingFraction - The faction of the buffers that have to be filled before the spilling thread actually begins spilling data to disk.

Throws:

IOException - Thrown, if an error occurs initializing the resources for external sorting.

MemoryAllocationException - Thrown, if not enough memory can be obtained from the memory manager to perform the sort.

CombiningUnilateralSortMerger

public CombiningUnilateralSortMerger(GroupCombineFunction<E,E> combineStub,
                                     MemoryManager memoryManager,
                                     IOManager ioManager,
                                     MutableObjectIterator<E> input,
                                     AbstractInvokable parentTask,
                                     TypeSerializerFactory<E> serializerFactory,
                                     TypeComparator<E> comparator,
                                     double memoryFraction,
                                     int numSortBuffers,
                                     int maxNumFileHandles,
                                     float startSpillingFraction,
                                     boolean objectReuseEnabled)
                              throws IOException,
                                     MemoryAllocationException

Creates a new sorter that reads the data from a given reader and provides an iterator returning that data in a sorted manner. The memory is divided among sort buffers, write buffers and read buffers automatically.

Parameters:

combineStub - The stub used to combine values with the same key.

memoryManager - The memory manager from which to allocate the memory.

ioManager - The I/O manager, which is used to write temporary files to disk.

input - The input that is sorted by this sorter.

parentTask - The parent task, which owns all resources used by this sorter.

serializerFactory - The type serializer.

comparator - The type comparator establishing the order relation.

memoryFraction - The fraction of memory dedicated to sorting, merging and I/O.

numSortBuffers - The number of distinct buffers to use creation of the initial runs.

maxNumFileHandles - The maximum number of files to be merged at once.

startSpillingFraction - The faction of the buffers that have to be filled before the spilling thread actually begins spilling data to disk.

Throws:

IOException - Thrown, if an error occurs initializing the resources for external sorting.

MemoryAllocationException - Thrown, if not enough memory can be obtained from the memory manager to perform the sort.

Method Detail

setUdfConfiguration

public void setUdfConfiguration(Configuration config)

getSpillingThread

protected UnilateralSortMerger.ThreadBase<E> getSpillingThread(ExceptionHandler<IOException> exceptionHandler,
                                                               UnilateralSortMerger.CircularQueues<E> queues,
                                                               AbstractInvokable parentTask,
                                                               MemoryManager memoryManager,
                                                               IOManager ioManager,
                                                               TypeSerializerFactory<E> serializerFactory,
                                                               TypeComparator<E> comparator,
                                                               List<MemorySegment> sortReadMemory,
                                                               List<MemorySegment> writeMemory,
                                                               int maxFileHandles)

Overrides:

getSpillingThread in class UnilateralSortMerger<E>