flink Datastream组装

transformation是flink中stream的静态对象，通过组装包含sink和source的transformation根据定义的代码可以组成stream的静态拓扑图，如下所示：

*   Source              Source
*      +                   +
*      |                   |
*      v                   v
*  Rebalance          HashPartition
*      +                   +
*      |                   |
*      |                   |
*      +------>Union<------+
*                +
*                |
*                v
*              Split
*                +
*                |
*                v
*              Select
*                +
*                v
*               Map
*                +
*                |
*                v
*              Sink

其中Source和SInk分别为拓扑图中的起点和终点，其中的spilt,select则是中间具体对数据进行操作的transformation。

同时DataStream类也是一个完成的一个stream数据抽象，其成员如下：

protected final StreamExecutionEnvironment environment;protected final StreamTransformation<T> transformation;

其中envirorment是整个流的上下文，其中的以数组的形式依次保存着之前的所有transformation，而transformation则是当前strean在组装过程中，逻辑上最后一个transformation。

下面也是一个简单的stream拓扑图例子：

*  Source              Source
*    +                   +
*    |                   |
*    |                   |
*    +------->Map<-------+
*              +
*              |
*              v
*             Sink

其中的map也是一个具体的streamTransformation，由DataStream调用map()方法具体加入到stream操作中。

public <R> SingleOutputStreamOperator<R> map(MapFunction<T, R> mapper) {TypeInformation<R> outType = TypeExtractor.getMapReturnTypes(clean(mapper), getType(),Utils.getCallLocationName(), true);return transform("Map", outType, new StreamMap<>(clean(mapper)));
}

在此处，当前一个DataStream调用map()方法的时候，首先需要得到当前DataStream的treanformation中输出的数据类型，根据其数据类型作为新的transformation的输入数据类型。

并需要具体实现map逻辑的MapFunction作为参数，否则没有意义。

之后将这一数据类型作为参数的一部分，在根据参数中的MapFunction生成新的StreamMap,也就是动态操作符，通过transform()方法，得到新的DataStream。

@PublicEvolving
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 MissingTypeInfotransformation.getOutputType();OneInputTransformation<T, R> resultTransform = new OneInputTransformation<>(this.transformation,operatorName,operator,outTypeInfo,environment.getParallelism());@SuppressWarnings({ "unchecked", "rawtypes" })SingleOutputStreamOperator<R> returnStream = new SingleOutputStreamOperator(environment, resultTransform);getExecutionEnvironment().addOperator(resultTransform);return returnStream;
}

在此处，由于是一个map操作，自然输入输出只需要一个，所以首先生成OneInputTransformation，这里的参数需要前一个transformation和当前的map操作和输入数据类型，以及当前上下文环境的并行度。

最后将生成的tramsform和上下文生成新的DataStream返回，作为新的DataStream进行下面的transform组装。

另一种例子是union操作，这一操作将会合并两个不同的DataStream。

@SafeVarargs
public final DataStream<T> union(DataStream<T>... streams) {List<StreamTransformation<T>> unionedTransforms = new ArrayList<>();unionedTransforms.add(this.transformation);for (DataStream<T> newStream : streams) {if (!getType().equals(newStream.getType())) {throw new IllegalArgumentException("Cannot union streams of different types: "+ getType() + " and " + newStream.getType());}unionedTransforms.add(newStream.getTransformation());}return new DataStream<>(this.environment, new UnionTransformation<>(unionedTransforms));
}
public UnionTransformation(List<StreamTransformation<T>> inputs) {super("Union", inputs.get(0).getOutputType(), inputs.get(0).getParallelism());for (StreamTransformation<T> input: inputs) {if (!input.getOutputType().equals(getOutputType())) {throw new UnsupportedOperationException("Type mismatch in input " + input);}}this.inputs = Lists.newArrayList(inputs);
}

此处会将所有需要合并的DataStream保存在数组当中，并生成一个新的UnionTransformation，其中将所有需要合并的DataStream数组参数作为例子保存在其中。新的UnionTransformation也将会作为新的DataStream的transform参数返回。

spilt和select同时使用可以达到分流的目的。

spilt()在被调用的时候需要一个OutputSelect并重写其select()方法，根据相应流入的数据返回不同的输出结果，而接下来再调用select()方法，并在参数中确定制定的返回结果，将会依照前者的输出结果定位到指定的流当中去。

flink Datastream组装相关推荐

flink DataStream API使用及原理
传统的大数据处理方式一般是批处理式的,也就是说,今天所收集的数据,我们明天再把今天收集到的数据算出来,以供大家使用,但是在很多情况下,数据的时效性对于业务的成败是非常关键的. Spark 和 Flin ...
Flink DataStream API 介绍
Flink DataStream API 介绍 StreamExecutionEnvironment #mermaid-svg-JKeWa22W2vWA4zBS {font-family:" ...
【基础】Flink -- DataStream API
Flink -- DataStream API 执行环境 Execution Environment 创建执行环境设置执行模式触发程序执行源算子 Source 从集合中读取数据从文件读取数据 ...
Flink专题四：Flink DataStream 窗口介绍及使用
由于工作需要最近学习flink 现记录下Flink介绍和实际使用过程这是flink系列的第四篇文章 Flink DataStream 窗口介绍及使用窗口介绍时间窗口翻滚窗口(数据以一个时间断为 ...
Apache Flink DataStream 编程全集
概述 Flink是构建在数据流之上的有状态计算的流计算框架,通常被人们理解为是第三代大数据分析方案. 第一代 - Hadoop的MapReduce计算(静态).Storm流计算(2014.9) :两套 ...
Flink DataStream iterate算子的简单使用
Flink DataStream iterate算子的简单使用由于DataStream程序可能永远不会完成,因此没有最大迭代次数.相反你需要指定流的哪个部分反馈到迭代,哪个部分使用split转换或转 ...
Apache Flink 零基础入门（十五）Flink DataStream编程（如何自定义DataSource）
数据源可以通过StreamExecutionEnvironment.addSource(sourceFunction)方式来创建,Flink也提供了一些内置的数据源方便使用,例如readTextFil ...
Flink DataStream 编程入门
流处理是 Flink 的核心,流处理的数据集用 DataStream 表示.数据流从可以从各种各样的数据源中创建(消息队列.Socket 和文件等),经过 DataStream 的各种 transf ...
[Flink]Flink DataStream window join 和interval join
目录 window join interval join window join 窗口连接把两个流中相同窗口通过一个键值连接起来.然后,两边的元素被传递到用户定义的JoinFunction或FlatJ ...

flink Datastream组装

flink Datastream组装相关推荐

最新文章

热门文章