先看例子，
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();