先看例子，
final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); DataStream<Tuple2<Long, Long>> stream = env.addSource(...); stream .keyBy(0) .timeWindow(Time.of(2500, MILLISECONDS), Time.of(500, MILLISECONDS)) .reduce(new SummingReducer()) .addSink(new SinkFunction<Tuple2<Long, Long>>() {...}); env.execute();
看出，和batch最大的不同是，这里是DataStream而不是DataSet；
/** * A DataStream represents a stream of elements of the same type. A DataStream * can be transformed into another DataStream by applying a transformation as * for example: * <ul> * <li>{@link DataStream#map}, * <li>{@link DataStream#filter}, or * </ul> * * @param <T> The type of the elements in this Stream */ public class DataStream<T> { protected final StreamExecutionEnvironment environment; protected final StreamTransformation<T> transformation; /** * Create a new {@link DataStream} in the given execution environment with * partitioning set to forward by default. * * @param environment The StreamExecutionEnvironment */ public DataStream(StreamExecutionEnvironment environment, StreamTransformation<T> transformation) { this.environment = Preconditions.checkNotNull(environment, "Execution Environment must not be null."); this.transformation = Preconditions.checkNotNull(transformation, "Stream Transformation must not be null."); } //DataStream上的各种操作。。。。。。 //map,reduce,keyby...... }
DataStream的核心，即
StreamTransformation<T> transformation; 如何产生data stream
StreamTransformation
对于StreamTransformation，表示一个用于create dataStream的operation；
并且不一定需要对应于一个实际的物理operation，可能只是个逻辑概念，比如下面的例子
/** * A {@code StreamTransformation} represents the operation that creates a * {@link org.apache.flink.streaming.api.datastream.DataStream}. Every * {@link org.apache.flink.streaming.api.datastream.DataStream} has an underlying * {@code StreamTransformation} that is the origin of said DataStream. * * <p> * API operations such as {@link org.apache.flink.streaming.api.datastream.DataStream#map} create * a tree of {@code StreamTransformation}s underneath. When the stream program is to be executed this * graph is translated to a {@link StreamGraph} using * {@link org.apache.flink.streaming.api.graph.StreamGraphGenerator}. * * <p> * A {@code StreamTransformation} does not necessarily correspond to a physical operation * at runtime. Some operations are only logical concepts. Examples of this are union, * split/select data stream, partitioning. * * <p> * The following graph of {@code StreamTransformations}: * * <pre>{@code * Source Source * + + * | | * v v * Rebalance HashPartition * + + * | | * | | * +------>Union<------+ * + * | * v * Split * + * | * v * Select * + * v * Map * + * | * v * Sink * }</pre> * * Would result in this graph of operations at runtime: * * <pre>{@code * Source Source * + + * | | * | | * +------->Map<-------+ * + * | * v * Sink * }</pre> * * The information about partitioning, union, split/select end up being encoded in the edges * that connect the sources to the map operation. * * @param <T> The type of the elements that result from this {@code StreamTransformation} */ public abstract class StreamTransformation<T>
对于StreamTransformation只定义了output，即该transform产生的result stream
这是抽象类无法直接用，transform产生stream的逻辑还是要封装在具体的operator中
通过下面的例子体会一下，transform和operator的区别，这里设计的有点绕
OneInputTransformation，在StreamTransformation基础上加上input
/** * This Transformation represents the application of a * {@link org.apache.flink.streaming.api.operators.OneInputStreamOperator} to one input * {@link org.apache.flink.streaming.api.transformations.StreamTransformation}. * * @param <IN> The type of the elements in the nput {@code StreamTransformation} * @param <OUT> The type of the elements that result from this {@code OneInputTransformation} */ public class OneInputTransformation<IN, OUT> extends StreamTransformation<OUT> { private final StreamTransformation<IN> input; private final OneInputStreamOperator<IN, OUT> operator; private KeySelector<IN, ?> stateKeySelector; private TypeInformation<?> stateKeyType; }
所以包含，
产生input stream的StreamTransformation<IN> input
以及通过input产生output的OneInputStreamOperator<IN, OUT> operator
同时也可以看下，
public class TwoInputTransformation<IN1, IN2, OUT> extends StreamTransformation<OUT> { private final StreamTransformation<IN1> input1; private final StreamTransformation<IN2> input2; private final TwoInputStreamOperator<IN1, IN2, OUT> operator; }
在看下SourceTransformation和SinkTransformation的对比，
public class SourceTransformation<T> extends StreamTransformation<T> { private final StreamSource<T> operator; } public class SinkTransformation<T> extends StreamTransformation<Object> { private final StreamTransformation<T> input; private final StreamSink<T> operator; }
比较容易理解transform的作用，
对于source，没有input，所以没有代表input的transformation
而对于sink，有input，但是sink的operator不是普通的streamOperator，是StreamSink，即流的终点
transform
这个函数的意思，用用户自定义的operator，将当前的Stream，转化为用户指定类型的Stream
/** * Method for passing user defined operators along with the type * information that will transform the DataStream. * * @param operatorName * name of the operator, for logging purposes * @param outTypeInfo * the output type of the operator * @param operator * the object containing the transformation logic * @param <R> * type of the return stream * @return the data stream constructed */ public <R> SingleOutputStreamOperator<R, ?> transform(String operatorName, TypeInformation<R> outTypeInfo, OneInputStreamOperator<T, R> operator) { // read the output type of the input Transform to coax out errors about MissingTypeInfo transformation.getOutputType(); OneInputTransformation<T, R> resultTransform = new OneInputTransformation<T, R>( this.transformation, operatorName, operator, outTypeInfo, environment.getParallelism()); @SuppressWarnings({ "unchecked", "rawtypes" }) SingleOutputStreamOperator<R, ?> returnStream = new SingleOutputStreamOperator(environment, resultTransform); getExecutionEnvironment().addOperator(resultTransform); return returnStream; }
所以参数为，
用户定义的: 输出的TypeInformation，以及OneInputStreamOperator
实现是，
创建OneInputTransformation，以this.transformation为input，以传入的operator为OneInputStreamOperator
所以通过resultTransform，就会将当前的stream转换为目的流
然后又封装一个SingleOutputStreamOperator，这是什么？
/** * The SingleOutputStreamOperator represents a user defined transformation * applied on a {@link DataStream} with one predefined output type. * * @param <T> The type of the elements in this Stream * @param <O> Type of the operator. */ public class SingleOutputStreamOperator<T, O extends SingleOutputStreamOperator<T, O>> extends DataStream<T> { protected SingleOutputStreamOperator(StreamExecutionEnvironment environment, StreamTransformation<T> transformation) { super(environment, transformation); } }
说白了，就是封装了一下用户定义的transformation
Flink这块代码的命名有点混乱，Operator，transformation，两个概念容易混
上面的例子，里面keyBy(0)
会产生
KeyedStream
对于keyedStream，关键的就是
keySelector和keyType，如何产生key以及key的类型
/** * 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. * * <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 KeyedStream<T, KEY> extends DataStream<T> { /** The key selector that can get the key by which the stream if partitioned from the elements */ private final KeySelector<T, KEY> keySelector; /** The type of the key by which the stream is partitioned */ private final TypeInformation<KEY> keyType; }
看下transform，在调用DataStream.transform的同时，设置keySelector和keyType
// ------------------------------------------------------------------------ // basic transformations // ------------------------------------------------------------------------ @Override public <R> SingleOutputStreamOperator<R, ?> transform(String operatorName, TypeInformation<R> outTypeInfo, OneInputStreamOperator<T, R> operator) { SingleOutputStreamOperator<R, ?> returnStream = super.transform(operatorName, outTypeInfo,operator); // inject the key selector and key type OneInputTransformation<T, R> transform = (OneInputTransformation<T, R>) returnStream.getTransformation(); transform.setStateKeySelector(keySelector); transform.setStateKeyType(keyType); return returnStream; }
KeyedStream很关键的是，作为一个到WindowedStream的过度，
所以提供一组生成Windowed的接口
// ------------------------------------------------------------------------ // Windowing // ------------------------------------------------------------------------ /** * Windows this {@code KeyedStream} into tumbling time windows. * * <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) { return window(TumblingTimeWindows.of(size)); }
WindowedStream
例子中
.timeWindow(Time.of(2500, MILLISECONDS), Time.of(500, MILLISECONDS))
/** * A {@code WindowedStream} represents a data stream where elements are grouped by * key, and for each key, the stream of elements is split into windows based on a * {@link org.apache.flink.streaming.api.windowing.assigners.WindowAssigner}. Window emission * is triggered based on a {@link org.apache.flink.streaming.api.windowing.triggers.Trigger}. * * <p> * The windows are conceptually evaluated for each key individually, meaning windows can trigger at * different points for each key. * * <p> * If an {@link Evictor} is specified it will be used to evict elements from the window after * evaluation was triggered by the {@code Trigger} but before the actual evaluation of the window. * When using an evictor window performance will degrade significantly, since * pre-aggregation of window results cannot be used. * * <p> * Note that the {@code WindowedStream} is purely and API construct, during runtime * the {@code WindowedStream} will be collapsed together with the * {@code KeyedStream} and the operation over the window into one single operation. * * @param <T> The type of elements in the stream. * @param <K> The type of the key by which elements are grouped. * @param <W> The type of {@code Window} that the {@code WindowAssigner} assigns the elements to. */ public class WindowedStream<T, K, W extends Window> { /** The keyed data stream that is windowed by this stream */ private final KeyedStream<T, K> input; /** The window assigner */ private final WindowAssigner<? super T, W> windowAssigner; /** The trigger that is used for window evaluation/emission. */ private Trigger<? super T, ? super W> trigger; /** The evictor that is used for evicting elements before window evaluation. */ private Evictor<? super T, ? super W> evictor;
可以看到WindowedStream没有直接继承自DataStream
而是以，KeyedStream作为他的input
当然window所必需的，WindowAssigner，Trigger和Evictor，也是不会少
继续例子， .reduce(new SummingReducer())
看看windowedStream的操作，reduce
/** * Applies a reduce function to the window. The window function is called for each evaluation * of the window for each key individually. The output of the reduce function is interpreted * as a regular non-windowed stream. * <p> * This window will try and pre-aggregate data as much as the window policies permit. For example, * tumbling time windows can perfectly pre-aggregate the data, meaning that only one element per * key is stored. Sliding time windows will pre-aggregate on the granularity of the slide interval, * so a few elements are stored per key (one per slide interval). * Custom windows may not be able to pre-aggregate, or may need to store extra values in an * aggregation tree. * * @param function The reduce function. * @return The data stream that is the result of applying the reduce function to the window. */ public SingleOutputStreamOperator<T, ?> reduce(ReduceFunction<T> function) { //clean the closure function = input.getExecutionEnvironment().clean(function); String opName = "TriggerWindow(" + windowAssigner + ", " + trigger + ", " + udfName + ")"; KeySelector<T, K> keySel = input.getKeySelector(); OneInputStreamOperator<T, T> operator; boolean setProcessingTime = input.getExecutionEnvironment().getStreamTimeCharacteristic() == TimeCharacteristic.ProcessingTime; if (evictor != null) { operator = new EvictingWindowOperator<>(windowAssigner, windowAssigner.getWindowSerializer(getExecutionEnvironment().getConfig()), keySel, input.getKeyType().createSerializer(getExecutionEnvironment().getConfig()), new HeapWindowBuffer.Factory<T>(), new ReduceWindowFunction<K, W, T>(function), trigger, evictor).enableSetProcessingTime(setProcessingTime); } else { operator = new WindowOperator<>(windowAssigner, windowAssigner.getWindowSerializer(getExecutionEnvironment().getConfig()), keySel, input.getKeyType().createSerializer(getExecutionEnvironment().getConfig()), new PreAggregatingHeapWindowBuffer.Factory<>(function), //PreAggre，即不会cache真实的element，而是直接存聚合过的值，这样比较节省空间 new ReduceWindowFunction<K, W, T>(function), trigger).enableSetProcessingTime(setProcessingTime); } return input.transform(opName, input.getType(), operator); }
关键就是根据是否有Evicting，选择创建不同的WindowOperator
然后调用input.transform，将windowedStream转换成SingleOutputStream，
这里input，即是keyedStream
// ------------------------------------------------------------------------ // basic transformations // ------------------------------------------------------------------------ @Override public <R> SingleOutputStreamOperator<R, ?> transform(String operatorName, TypeInformation<R> outTypeInfo, OneInputStreamOperator<T, R> operator) { SingleOutputStreamOperator<R, ?> returnStream = super.transform(operatorName, outTypeInfo,operator); // inject the key selector and key type OneInputTransformation<T, R> transform = (OneInputTransformation<T, R>) returnStream.getTransformation(); transform.setStateKeySelector(keySelector); transform.setStateKeyType(keyType); return returnStream; }
可以看到这里的参数是OneInputStreamOperator，而WindowOperator其实是实现了该interface的，
可以看到，对于OneInputStreamOperator而言，我们只需要实现，processElement和processWatermark两个接口，侧重如何处理input element
/** * Interface for stream operators with one input. Use * {@link org.apache.flink.streaming.api.operators.AbstractStreamOperator} as a base class if * you want to implement a custom operator. * * @param <IN> The input type of the operator * @param <OUT> The output type of the operator */ public interface OneInputStreamOperator<IN, OUT> extends StreamOperator<OUT> { /** * Processes one element that arrived at this operator. * This method is guaranteed to not be called concurrently with other methods of the operator. */ void processElement(StreamRecord<IN> element) throws Exception; /** * Processes a {@link Watermark}. * This method is guaranteed to not be called concurrently with other methods of the operator. * * @see org.apache.flink.streaming.api.watermark.Watermark */ void processWatermark(Watermark mark) throws Exception; }
继续调用，super.transform，即DataStream的transform
例子最后，
.addSink(new SinkFunction<Tuple2<Long, Long>>() {...});
实际是调用，
SingleOutputStreamOperator.addSink，即DataStream.addSink
/** * Adds the given sink to this DataStream. Only streams with sinks added * will be executed once the {@link StreamExecutionEnvironment#execute()} * method is called. * * @param sinkFunction * The object containing the sink's invoke function. * @return The closed DataStream. */ public DataStreamSink<T> addSink(SinkFunction<T> sinkFunction) { StreamSink<T> sinkOperator = new StreamSink<T>(clean(sinkFunction)); DataStreamSink<T> sink = new DataStreamSink<T>(this, sinkOperator); getExecutionEnvironment().addOperator(sink.getTransformation()); return sink; }
SinkFunction结构，
public interface SinkFunction<IN> extends Function, Serializable { /** * Function for standard sink behaviour. This function is called for every record. * * @param value The input record. * @throws Exception */ void invoke(IN value) throws Exception; }
StreamSink，即是OneInputStreamOperator，所以主要是processElement接口
public class StreamSink<IN> extends AbstractUdfStreamOperator<Object, SinkFunction<IN>> implements OneInputStreamOperator<IN, Object> { public StreamSink(SinkFunction<IN> sinkFunction) { super(sinkFunction); chainingStrategy = ChainingStrategy.ALWAYS; } @Override public void processElement(StreamRecord<IN> element) throws Exception { userFunction.invoke(element.getValue()); } @Override public void processWatermark(Watermark mark) throws Exception { // ignore it for now, we are a sink, after all } }
DataStreamSink，就是对SinkTransformation的封装
/** * A Stream Sink. This is used for emitting elements from a streaming topology. * * @param <T> The type of the elements in the Stream */ public class DataStreamSink<T> { SinkTransformation<T> transformation; @SuppressWarnings("unchecked") protected DataStreamSink(DataStream<T> inputStream, StreamSink<T> operator) { this.transformation = new SinkTransformation<T>(inputStream.getTransformation(), "Unnamed", operator, inputStream.getExecutionEnvironment().getParallelism()); } }
最终，
把SinkTransformation加入 List<StreamTransformation<?>> transformations
最后走到，env.execute();