flink-issues mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From ktzoumas <...@git.apache.org>
Subject [GitHub] flink pull request: Stream API Refactoring
Date Mon, 05 Oct 2015 09:34:11 GMT
Github user ktzoumas commented on a diff in the pull request:

    https://github.com/apache/flink/pull/1215#discussion_r41125422
  
    --- Diff: flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/datastream/KeyedStream.java
---
    @@ -24,49 +24,169 @@
     import org.apache.flink.api.java.functions.KeySelector;
     import org.apache.flink.api.java.typeutils.TypeExtractor;
     import org.apache.flink.streaming.api.functions.aggregation.AggregationFunction;
    -import org.apache.flink.streaming.api.functions.aggregation.AggregationFunction.AggregationType;
     import org.apache.flink.streaming.api.functions.aggregation.ComparableAggregator;
     import org.apache.flink.streaming.api.functions.aggregation.SumAggregator;
    +import org.apache.flink.streaming.api.functions.sink.SinkFunction;
    +import org.apache.flink.streaming.api.operators.OneInputStreamOperator;
     import org.apache.flink.streaming.api.operators.StreamGroupedFold;
     import org.apache.flink.streaming.api.operators.StreamGroupedReduce;
    +import org.apache.flink.streaming.api.transformations.OneInputTransformation;
    +import org.apache.flink.streaming.api.transformations.PartitionTransformation;
    +import org.apache.flink.streaming.api.windowing.assigners.SlidingProcessingTimeWindows;
    +import org.apache.flink.streaming.api.windowing.assigners.SlidingTimeWindows;
    +import org.apache.flink.streaming.api.windowing.assigners.TumblingProcessingTimeWindows;
    +import org.apache.flink.streaming.api.windowing.assigners.TumblingTimeWindows;
    +import org.apache.flink.streaming.api.windowing.assigners.WindowAssigner;
    +import org.apache.flink.streaming.api.windowing.time.AbstractTime;
    +import org.apache.flink.streaming.api.windowing.time.EventTime;
    +import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
    +import org.apache.flink.streaming.api.windowing.windows.Window;
    +import org.apache.flink.streaming.runtime.partitioner.HashPartitioner;
    +import org.apache.flink.streaming.runtime.partitioner.StreamPartitioner;
     
     /**
    - * A GroupedDataStream represents a {@link DataStream} which has been
    - * partitioned by the given {@link KeySelector}. Operators like {@link #reduce},
    - * {@link #fold} etc. can be applied on the {@link GroupedDataStream} to
    - * get additional functionality by the grouping.
    + * A {@code KeyedStream} represents a {@link DataStream} on which operator state is
    + * partitioned by key using a provided {@link KeySelector}. Typical operations supported
by a
    + * {@code DataStream} are also possible on a {@code KeyedStream}, with the exception
of
    + * partitioning methods such as shuffle, forward and keyBy.
      *
    - * @param <T> The type of the elements in the Grouped Stream.
    + * <p>
    + * Reduce-style operations, such as {@link #reduce}, {@link #sum} and {@link #fold} work
on elements
    + * that have the same key.
    + *
    + * @param <T> The type of the elements in the Keyed Stream.
      * @param <KEY> The type of the key in the Keyed Stream.
      */
    -public class GroupedDataStream<T, KEY> extends KeyedDataStream<T, KEY> {
    +public class KeyedStream<T, KEY> extends DataStream<T> {
    +	
    +	protected final KeySelector<T, KEY> keySelector;
    +
    +	/**
    +	 * Creates a new {@link KeyedStream} using the given {@link KeySelector}
    +	 * to partition operator state by key.
    +	 * 
    +	 * @param dataStream
    +	 *            Base stream of data
    +	 * @param keySelector
    +	 *            Function for determining state partitions
    +	 */
    +	public KeyedStream(DataStream<T> dataStream, KeySelector<T, KEY> keySelector)
{
    +		super(dataStream.getExecutionEnvironment(), new PartitionTransformation<>(dataStream.getTransformation(),
new HashPartitioner<>(keySelector)));
    +		this.keySelector = keySelector;
    +	}
    +
    +	
    +	public KeySelector<T, KEY> getKeySelector() {
    +		return this.keySelector;
    +	}
    +
    +	
    +	@Override
    +	protected DataStream<T> setConnectionType(StreamPartitioner<T> partitioner)
{
    +		throw new UnsupportedOperationException("Cannot override partitioning for KeyedStream.");
    +	}
    +
    +	
    +	@Override
    +	public <R> SingleOutputStreamOperator<R, ?> transform(String operatorName,
    +			TypeInformation<R> outTypeInfo, OneInputStreamOperator<T, R> operator)
{
    +
    +		SingleOutputStreamOperator<R, ?> returnStream = super.transform(operatorName,
outTypeInfo,operator);
    +
    +		((OneInputTransformation<T, R>) returnStream.getTransformation()).setStateKeySelector(keySelector);
    +		return returnStream;
    +	}
    +
    +	
    +	
    +	@Override
    +	public DataStreamSink<T> addSink(SinkFunction<T> sinkFunction) {
    +		DataStreamSink<T> result = super.addSink(sinkFunction);
    +		result.getTransformation().setStateKeySelector(keySelector);
    +		return result;
    +	}
    +	
    +	// ------------------------------------------------------------------------
    +	//  Windowing
    +	// ------------------------------------------------------------------------
     
     	/**
    -	 * Creates a new {@link GroupedDataStream}, group inclusion is determined using
    -	 * a {@link KeySelector} on the elements of the {@link DataStream}.
    +	 * Windows this {@code KeyedStream} into tumbling time windows.
     	 *
    -	 * @param dataStream Base stream of data
    -	 * @param keySelector Function for determining group inclusion
    +	 * <p>
    +	 * This is a shortcut for either {@code .window(TumblingTimeWindows.of(size))} or
    +	 * {@code .window(TumblingProcessingTimeWindows.of(size))} depending on the time characteristic
    +	 * set using
    +	 * {@link org.apache.flink.streaming.api.environment.StreamExecutionEnvironment#setStreamTimeCharacteristic(org.apache.flink.streaming.api.TimeCharacteristic)}
    +	 *
    +	 * @param size The size of the window.
    +	 */
    +	public WindowedStream<T, KEY, TimeWindow> timeWindow(AbstractTime size) {
    +		AbstractTime actualSize = size.makeSpecificBasedOnTimeCharacteristic(environment.getStreamTimeCharacteristic());
    +
    +		if (actualSize instanceof EventTime) {
    +			return window(TumblingTimeWindows.of(actualSize));
    +		} else {
    +			return window(TumblingProcessingTimeWindows.of(actualSize));
    +		}
    +	}
    +
    +	/**
    +	 * Windows this {@code KeyedStream} into sliding time windows.
    +	 *
    +	 * <p>
    +	 * This is a shortcut for either {@code .window(SlidingTimeWindows.of(size, slide))}
or
    +	 * {@code .window(SlidingProcessingTimeWindows.of(size, slide))} depending on the time
characteristic
    +	 * set using
    +	 * {@link org.apache.flink.streaming.api.environment.StreamExecutionEnvironment#setStreamTimeCharacteristic(org.apache.flink.streaming.api.TimeCharacteristic)}
    +	 *
    +	 * @param size The size of the window.
     	 */
    -	public GroupedDataStream(DataStream<T> dataStream, KeySelector<T, KEY> keySelector)
{
    -		super(dataStream, keySelector);
    +	public WindowedStream<T, KEY, TimeWindow> timeWindow(AbstractTime size, AbstractTime
slide) {
    +		AbstractTime actualSize = size.makeSpecificBasedOnTimeCharacteristic(environment.getStreamTimeCharacteristic());
    +		AbstractTime actualSlide = slide.makeSpecificBasedOnTimeCharacteristic(environment.getStreamTimeCharacteristic());
    +
    +		if (actualSize instanceof EventTime) {
    +			return window(SlidingTimeWindows.of(actualSize, actualSlide));
    +		} else {
    +			return window(SlidingProcessingTimeWindows.of(actualSize, actualSlide));
    +		}
    +	}
    +
    +	/**
    +	 * Windows this data stream to a {@code WindowedStream}, which evaluates windows
    +	 * over a key grouped stream. Elements are put into windows by a {@link WindowAssigner}.
The
    +	 * grouping of elements is done both by key and by window.
    +	 *
    +	 * <p>
    +	 * A {@link org.apache.flink.streaming.api.windowing.triggers.Trigger} can be defined
to specify
    +	 * when windows are evaluated. However, {@code WindowAssigners} have a default {@code
Trigger}
    +	 * that is used if a {@code Trigger} is not specified.
    +	 *
    +	 * @param assigner The {@code WindowAssigner} that assigns elements to windows.
    +	 * @return The trigger windows data stream.
    +	 */
    +	public <W extends Window> WindowedStream<T, KEY, W> window(WindowAssigner<?
super T, W> assigner) {
    +		return new WindowedStream<>(this, assigner);
     	}
     
    +	// ------------------------------------------------------------------------
    +	//  Non-Windowed aggregation operations
    +	// ------------------------------------------------------------------------
     
     	/**
     	 * Applies a reduce transformation on the grouped data stream grouped on by
     	 * the given key position. The {@link ReduceFunction} will receive input
     	 * values based on the key value. Only input values with the same key will
     	 * go to the same reducer.
    -	 * 
    +	 *
     	 * @param reducer
     	 *            The {@link ReduceFunction} that will be called for every
     	 *            element of the input values with the same key.
     	 * @return The transformed DataStream.
     	 */
     	public SingleOutputStreamOperator<T, ?> reduce(ReduceFunction<T> reducer)
{
    -		return transform("Grouped Reduce", getType(), new StreamGroupedReduce<T>(
    -				clean(reducer), keySelector));
    +		return transform("Grouped Reduce", getType(), new StreamGroupedReduce<>(clean(reducer),
keySelector));
    --- End diff --
    
    "Grouped Reduce" or simply "Reduce"?


---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---

Mime
View raw message