Flink - DataStream

蔚落 2021-09-19 15:04 419阅读 0赞

先看例子，

![复制代码][copycode.gif]

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();

![复制代码][copycode.gif]

看出，和batch最大的不同是，这里是DataStream而不是DataSet；

![复制代码][copycode.gif]

/**
 * 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......
 }

![复制代码][copycode.gif]

DataStream的核心，即

StreamTransformation<T> transformation; 如何产生data stream

StreamTransformation

对于StreamTransformation，表示一个用于create dataStream的operation；   
并且不一定需要对应于一个实际的物理operation，可能只是个逻辑概念，比如下面的例子

![复制代码][copycode.gif]

/**
 * 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.
 *
 * 
 * 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}.
 *
 * 
 * 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.
 *
 * 
 * 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>

![复制代码][copycode.gif]

对于StreamTransformation只定义了output，即该transform产生的result stream   
这是抽象类无法直接用，transform产生stream的逻辑还是要封装在具体的operator中

通过下面的例子体会一下，transform和operator的区别，这里设计的有点绕

OneInputTransformation，在StreamTransformation基础上加上input

![复制代码][copycode.gif]

/**
 * 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;
 }

![复制代码][copycode.gif]

所以包含，  
产生input stream的StreamTransformation<IN> input  
以及通过input产生output的OneInputStreamOperator<IN, OUT> operator

同时也可以看下，

![复制代码][copycode.gif]

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;
 }

![复制代码][copycode.gif]

在看下SourceTransformation和SinkTransformation的对比，

![复制代码][copycode.gif]

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;
 }

![复制代码][copycode.gif]

比较容易理解transform的作用，   
对于source，没有input，所以没有代表input的transformation   
而对于sink，有input，但是sink的operator不是普通的streamOperator，是StreamSink，即流的终点

transform

这个函数的意思，用用户自定义的operator，将当前的Stream，转化为用户指定类型的Stream

![复制代码][copycode.gif]

/**
 * 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;
 }

![复制代码][copycode.gif]

所以参数为，

用户定义的: 输出的TypeInformation，以及OneInputStreamOperator

实现是，

创建OneInputTransformation，以this.transformation为input，以传入的operator为OneInputStreamOperator   
所以通过resultTransform，就会将当前的stream转换为目的流

然后又封装一个SingleOutputStreamOperator，这是什么？

![复制代码][copycode.gif]

/**
 * 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);
 }
 }

![复制代码][copycode.gif]

说白了，就是封装了一下用户定义的transformation

Flink这块代码的命名有点混乱，Operator，transformation，两个概念容易混

上面的例子，里面keyBy(0)

会产生

KeyedStream

对于keyedStream，关键的就是

keySelector和keyType，如何产生key以及key的类型

![复制代码][copycode.gif]

/**
 * 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.
 *
 * 
 * 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;
 }

![复制代码][copycode.gif]

看下transform，在调用DataStream.transform的同时，设置keySelector和keyType

![复制代码][copycode.gif]

// ------------------------------------------------------------------------
 // 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;
 }

![复制代码][copycode.gif]

KeyedStream很关键的是，作为一个到WindowedStream的过度，

所以提供一组生成Windowed的接口

![复制代码][copycode.gif]

// ------------------------------------------------------------------------
 // Windowing
 // ------------------------------------------------------------------------
 
 /**
 * Windows this {
 @code KeyedStream} into tumbling time windows.
 *
 * 
 * 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));
 }

![复制代码][copycode.gif]

### WindowedStream ###

例子中   
.timeWindow(Time.of(2500, MILLISECONDS), Time.of(500, MILLISECONDS))

![复制代码][copycode.gif]

/**
 * 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}.
 *
 * 
 * The windows are conceptually evaluated for each key individually, meaning windows can trigger at
 * different points for each key.
 *
 * 
 * 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.
 *
 * 
 * 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;

![复制代码][copycode.gif]

可以看到WindowedStream没有直接继承自DataStream

而是以，KeyedStream作为他的input

当然window所必需的，WindowAssigner，Trigger和Evictor，也是不会少

继续例子， .reduce(new SummingReducer())

看看windowedStream的操作，reduce

![复制代码][copycode.gif]

/**
 * 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.
 * 
 * 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);
 }

![复制代码][copycode.gif]

关键就是根据是否有Evicting，选择创建不同的WindowOperator

然后调用input.transform，将windowedStream转换成SingleOutputStream，

这里input，即是keyedStream

![复制代码][copycode.gif]

可以看到这里的参数是OneInputStreamOperator，而WindowOperator其实是实现了该interface的，

可以看到，对于OneInputStreamOperator而言，我们只需要实现，processElement和processWatermark两个接口，侧重如何处理input element

![复制代码][copycode.gif]

/**
 * 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;
 }

![复制代码][copycode.gif]

继续调用，super.transform，即DataStream的transform

例子最后，

.addSink(new SinkFunction<Tuple2<Long, Long>>() \{...\});

实际是调用，

SingleOutputStreamOperator.addSink，即DataStream.addSink

![复制代码][copycode.gif]

/**
 * 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;
 }

![复制代码][copycode.gif]

SinkFunction结构，

![复制代码][copycode.gif]

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;
 }

![复制代码][copycode.gif]

StreamSink，即是OneInputStreamOperator，所以主要是processElement接口

![复制代码][copycode.gif]

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
 }
 }

![复制代码][copycode.gif]

DataStreamSink，就是对SinkTransformation的封装

![复制代码][copycode.gif]

/**
 * 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());
 }
 }

![复制代码][copycode.gif]

最终，

把SinkTransformation加入 List<StreamTransformation<?>> transformations

最后走到，env.execute();

[copycode.gif]: /images/20210726/f8a3ba8158c04d3d9bbc2607703bbe21.png