@Internal public class KeyedPartitionWindowedStream<T,KEY> extends Object implements PartitionWindowedStream<T>
KeyedPartitionWindowedStream
represents a data stream that collects all records with the
same key separately into a full window.Constructor and Description |
---|
KeyedPartitionWindowedStream(StreamExecutionEnvironment environment,
KeyedStream<T,KEY> input) |
Modifier and Type | Method and Description |
---|---|
<ACC,R> SingleOutputStreamOperator<R> |
aggregate(AggregateFunction<T,ACC,R> aggregateFunction)
Applies an aggregate transformation on the records of the window.
|
<R> SingleOutputStreamOperator<R> |
mapPartition(MapPartitionFunction<T,R> mapPartitionFunction)
Process the records of the window by
MapPartitionFunction . |
SingleOutputStreamOperator<T> |
reduce(ReduceFunction<T> reduceFunction)
Applies a reduce transformation on the records of the window.
|
SingleOutputStreamOperator<T> |
sortPartition(int field,
Order order)
Sorts the records of the window on the specified field in the specified order.
|
<K> SingleOutputStreamOperator<T> |
sortPartition(KeySelector<T,K> keySelector,
Order order)
Sorts the records according to a
KeySelector in the specified order. |
SingleOutputStreamOperator<T> |
sortPartition(String field,
Order order)
Sorts the records of the window on the specified field in the specified order.
|
public KeyedPartitionWindowedStream(StreamExecutionEnvironment environment, KeyedStream<T,KEY> input)
public <R> SingleOutputStreamOperator<R> mapPartition(MapPartitionFunction<T,R> mapPartitionFunction)
PartitionWindowedStream
MapPartitionFunction
.mapPartition
in interface PartitionWindowedStream<T>
R
- The type of map partition result.mapPartitionFunction
- The map partition function.public SingleOutputStreamOperator<T> reduce(ReduceFunction<T> reduceFunction)
PartitionWindowedStream
reduce
in interface PartitionWindowedStream<T>
reduceFunction
- The reduce function.public <ACC,R> SingleOutputStreamOperator<R> aggregate(AggregateFunction<T,ACC,R> aggregateFunction)
PartitionWindowedStream
aggregate
in interface PartitionWindowedStream<T>
ACC
- The type of accumulator in aggregate function.R
- The type of aggregate function result.aggregateFunction
- The aggregate function.public SingleOutputStreamOperator<T> sortPartition(int field, Order order)
PartitionWindowedStream
Tuple
.
This operator will use managed memory for the sort.For NonKeyedPartitionWindowedStream
, the managed memory size can be set by ExecutionOptions.SORT_PARTITION_MEMORY
. For KeyedPartitionWindowedStream
, the
managed memory size can be set by ExecutionOptions.SORT_KEYED_PARTITION_MEMORY
.
sortPartition
in interface PartitionWindowedStream<T>
field
- The field 1-based index on which records is sorted.order
- The order in which records is sorted.public SingleOutputStreamOperator<T> sortPartition(String field, Order order)
PartitionWindowedStream
PojoTypeInfo
. A type is considered a Flink POJO type, if
it fulfills the conditions below.
This operator will use managed memory for the sort.For NonKeyedPartitionWindowedStream
, the managed memory size can be set by ExecutionOptions.SORT_PARTITION_MEMORY
. For KeyedPartitionWindowedStream
, the
managed memory size can be set by ExecutionOptions.SORT_KEYED_PARTITION_MEMORY
.
sortPartition
in interface PartitionWindowedStream<T>
field
- The field expression referring to the field on which records is sorted.order
- The order in which records is sorted.public <K> SingleOutputStreamOperator<T> sortPartition(KeySelector<T,K> keySelector, Order order)
PartitionWindowedStream
KeySelector
in the specified order.
This operator will use managed memory for the sort.For NonKeyedPartitionWindowedStream
, the managed memory size can be set by ExecutionOptions.SORT_PARTITION_MEMORY
. For KeyedPartitionWindowedStream
, the
managed memory size can be set by ExecutionOptions.SORT_KEYED_PARTITION_MEMORY
.
sortPartition
in interface PartitionWindowedStream<T>
keySelector
- The key selector to extract key from the records for sorting.order
- The order in which records is sorted.Copyright © 2014–2024 The Apache Software Foundation. All rights reserved.