4.1、Flink流处理（Stream API）- Operators（操作数据流）

左手的ㄟ右手 2021-10-19 08:26 314阅读 0赞

**目录**

DataStream Transformations（流数据转换）

物理分区

任务链接和资源组

--------------------

Operators将一个或多个数据流转换为一个新的数据流。程序可以将多个转换组合成复杂的数据流拓扑。

描述基本转换、应用这些转换之后的有效物理分区以及对Flink操作符链接的理解。

# DataStream Transformations（流数据转换） #

<table> 
 <tbody> 
 <tr> 
 <td style="width:204px;"> 转换方法 </td> 
 <td style="width:645px;"> 描述及代码样例 </td> 
 </tr> 
 <tr> 
 <td style="width:204px;">Map DataStream → DataStream</td> 
 <td style="width:645px;"> 获取一个元素并生成一个元素。一个map函数，例子：使输入流的值加倍： <pre><code>DataStream&lt;Integer&gt; dataStream = //...
dataStream.map(new MapFunction&lt;Integer, Integer&gt;() {
 @Override
 public Integer map(Integer value) throws Exception {
 return 2 * value;
 }
});
 </code></pre> &nbsp; </td> 
 </tr> 
 <tr> 
 <td style="width:204px;">FlatMap DataStream → DataStream</td> 
 <td style="width:645px;"> 获取一个元素并生成零个、一个或多个元素。一种将句子分割成单词的平面映射函数： <pre><code>dataStream.flatMap(new FlatMapFunction&lt;String, String&gt;() {
 @Override
 public void flatMap(String value, Collector&lt;String&gt; out)
 throws Exception {
 for(String word: value.split(" ")){
 out.collect(word);
 }
 }
});
 </code></pre> &nbsp; </td> 
 </tr> 
 <tr> 
 <td style="width:204px;">Filter DataStream → DataStream</td> 
 <td style="width:645px;"> 为每个元素计算布尔函数，并保留函数返回true的元素。过滤掉零值的过滤器： <pre><code>dataStream.filter(new FilterFunction&lt;Integer&gt;() {
 @Override
 public boolean filter(Integer value) throws Exception {
 return value != 0;
 }
});
 </code></pre> &nbsp; </td> 
 </tr> 
 <tr> 
 <td style="width:204px;">KeyBy DataStream → KeyedStream</td> 
 <td style="width:645px;"> 逻辑上将流划分为不相交的分区。所有具有相同key的记录都被分配到同一个分区。在内部，keyBy()是通过哈希分区实现的。有不同的方法来指定键。 这个转换返回一个KeyedStream，它是使用key状态所必需的。 <pre><code>dataStream.keyBy("someKey") // Key by field "someKey"
dataStream.keyBy(0) // Key by the first element of a Tuple
 </code></pre> 注意：下面的类型不能作为key： 
 <ol> 
 <li>没有重写hashCode的POJO对象，依赖于Object.hashCode()实现</li> 
 <li>任意类型的数组</li> 
 </ol></td> 
 </tr> 
 <tr> 
 <td style="width:204px;">Reduce KeyedStream → DataStream</td> 
 <td style="width:645px;"> 基于keyBy()，将当前元素与最后一个缩减值组合并返回新值。 <pre><code>keyedStream.reduce(new ReduceFunction&lt;Integer&gt;() {
 @Override
 public Integer reduce(Integer value1, Integer value2)
 throws Exception {
 return value1 + value2;
 }
});
 </code></pre> &nbsp; </td> 
 </tr> 
 <tr> 
 <td style="width:204px;">Fold KeyedStream → DataStream</td> 
 <td style="width:645px;"> 基于keyBy()，将当前元素与最后一个折叠值组合并返回新值。 应用于序列(1,2,3,4,5)时，返回序列“start-1”，“start-1-2”，“start-1-2-3”，… <pre><code>DataStream&lt;String&gt; result =
 keyedStream.fold("start", new FoldFunction&lt;Integer, String&gt;() {
 @Override
 public String fold(String current, Integer value) {
 return current + "-" + value;
 }
 });
 </code></pre> &nbsp; </td> 
 </tr> 
 <tr> 
 <td style="width:204px;">Aggregations KeyedStream → DataStream</td> 
 <td style="width:645px;"> 基于keyBy()，min和minBy的区别在于，min返回最小值，而minBy返回该字段中具有最小值的元素(max和maxBy也是如此)。 <pre><code>keyedStream.sum(0);
keyedStream.sum("key");
keyedStream.min(0);
keyedStream.min("key");
keyedStream.max(0);
keyedStream.max("key");
keyedStream.minBy(0);
keyedStream.minBy("key");
keyedStream.maxBy(0);
keyedStream.maxBy("key");</code></pre> &nbsp; </td> 
 </tr> 
 <tr> 
 <td style="width:204px;">Window KeyedStream → WindowedStream</td> 
 <td style="width:645px;"> Windows可以在已经分区的KeyedStreams上定义。Windows根据某些特性(例如，最近5秒内到达的数据)对每个键中的数据进行分组。Window的详解//TODO <pre><code>dataStream.keyBy(0).window(TumblingEventTimeWindows.of(Time.seconds(5))); // Last 5 seconds of data
 </code></pre> &nbsp; </td> 
 </tr> 
 <tr> 
 <td style="width:204px;">WindowAll DataStream → AllWindowedStream</td> 
 <td style="width:645px;"> 可以在常规数据流上定义。Windows根据一些特性(例如，最近5秒内到达的数据)对所有流事件进行分组。 注意：在大多案例中，这是一个非并行转换。所有记录将在windowAll操作符的一个任务中收集。 <pre><code>dataStream.windowAll(TumblingEventTimeWindows.of(Time.seconds(5))); // Last 5 seconds of data
 </code></pre> &nbsp; </td> 
 </tr> 
 <tr> 
 <td style="width:204px;">Window Apply WindowedStream → DataStream AllWindowedStream → DataStream</td> 
 <td style="width:645px;"> 将一个通用函数应用于整个窗口。下面是一个手动汇总窗口元素的函数。 注意：如果正在使用windowAll转换，则需要使用AllWindowFunction。 <pre><code>windowedStream.apply (new WindowFunction&lt;Tuple2&lt;String,Integer&gt;, Integer, Tuple, Window&gt;() {
 public void apply (Tuple tuple,
 Window window,
 Iterable&lt;Tuple2&lt;String, Integer&gt;&gt; values,
 Collector&lt;Integer&gt; out) throws Exception {
 int sum = 0;
 for (value t: values) {
 sum += t.f1;
 }
 out.collect (new Integer(sum));
 }
});

// applying an AllWindowFunction on non-keyed window stream
allWindowedStream.apply (new AllWindowFunction&lt;Tuple2&lt;String,Integer&gt;, Integer, Window&gt;() {
 public void apply (Window window,
 Iterable&lt;Tuple2&lt;String, Integer&gt;&gt; values,
 Collector&lt;Integer&gt; out) throws Exception {
 int sum = 0;
 for (value t: values) {
 sum += t.f1;
 }
 out.collect (new Integer(sum));
 }
});
 </code></pre> &nbsp; </td> 
 </tr> 
 <tr> 
 <td style="width:204px;">Window Reduce WindowedStream → DataStream</td> 
 <td style="width:645px;"> 将reduce函数应用于窗口并返回reduce后的值。 <pre><code>windowedStream.reduce (new ReduceFunction&lt;Tuple2&lt;String,Integer&gt;&gt;() {
 public Tuple2&lt;String, Integer&gt; reduce(Tuple2&lt;String, Integer&gt; value1, Tuple2&lt;String, Integer&gt; value2) throws Exception {
 return new Tuple2&lt;String,Integer&gt;(value1.f0, value1.f1 + value2.f1);
 }
});
 </code></pre> &nbsp; </td> 
 </tr> 
 <tr> 
 <td style="width:204px;">Window Fold WindowedStream → DataStream</td> 
 <td style="width:645px;"> 将fold函数应用于窗口并返回fold后的值。将示例函数应用于序列(1,2,3,4,5)时，将序列折叠成字符串“start 1-2-3-4-5”： <pre><code>windowedStream.fold("start", new FoldFunction&lt;Integer, String&gt;() {
 public String fold(String current, Integer value) {
 return current + "-" + value;
 }
});
 </code></pre> &nbsp; </td> 
 </tr> 
 <tr> 
 <td style="width:204px;">Aggregations on windows WindowedStream → DataStream</td> 
 <td style="width:645px;"> 基于window()，in和minBy的区别在于，min返回最小值，而minBy返回该字段中具有最小值的元素(max和maxBy也是如此)。 <pre><code>windowedStream.sum(0);
windowedStream.sum("key");
windowedStream.min(0);
windowedStream.min("key");
windowedStream.max(0);
windowedStream.max("key");
windowedStream.minBy(0);
windowedStream.minBy("key");
windowedStream.maxBy(0);
windowedStream.maxBy("key");
 </code></pre> &nbsp; </td> 
 </tr> 
 <tr> 
 <td style="width:204px;">Union DataStream* → DataStream</td> 
 <td style="width:645px;"> 两个或多个数据流的联合，创建一个包含来自所有流的所有元素的新流。注意：如果将一个数据流与它自己相结合，将得到结果流中的每个元素两次。 <pre><code>dataStream.union(otherStream1, otherStream2, ...);</code></pre> &nbsp; </td> 
 </tr> 
 <tr> 
 <td style="width:204px;">Window Join DataStream,DataStream → DataStream</td> 
 <td style="width:645px;"> 连接给定键和公共窗口上的两个数据流。 <pre><code>dataStream.join(otherStream)
 .where(&lt;key selector&gt;).equalTo(&lt;key selector&gt;)
 .window(TumblingEventTimeWindows.of(Time.seconds(3)))
 .apply (new JoinFunction () {...});
 </code></pre> &nbsp; </td> 
 </tr> 
 <tr> 
 <td style="width:204px;">Interval Join KeyedStream,KeyedStream → DataStream</td> 
 <td style="width:645px;"> 将两个键流的两个元素e1和e2用一个公共键在给定的时间间隔内连接起来，这样 e1.timestamp + lowerBound &lt;= e2.timestamp &lt;= e1.timestamp + upperBound <pre><code>// this will join the two streams so that
// key1 == key2 &amp;&amp; leftTs - 2 &lt; rightTs &lt; leftTs + 2
keyedStream.intervalJoin(otherKeyedStream)
 .between(Time.milliseconds(-2), Time.milliseconds(2)) // lower and upper bound
 .upperBoundExclusive(true) // optional
 .lowerBoundExclusive(true) // optional
 .process(new IntervalJoinFunction() {...});
 </code></pre> &nbsp; </td> 
 </tr> 
 <tr> 
 <td style="width:204px;">Window CoGroup DataStream,DataStream → DataStream</td> 
 <td style="width:645px;"> 将给定键和公共窗口上的两个数据流组合在一起。 <pre><code>dataStream.coGroup(otherStream)
 .where(0).equalTo(1)
 .window(TumblingEventTimeWindows.of(Time.seconds(3)))
 .apply (new CoGroupFunction () {...});
 </code></pre> &nbsp; </td> 
 </tr> 
 <tr> 
 <td style="width:204px;">Connect DataStream,DataStream → ConnectedStreams</td> 
 <td style="width:645px;"> 连接两个保留类型的数据流。允许两个流之间共享状态的连接。 <pre><code>DataStream&lt;Integer&gt; someStream = //...
DataStream&lt;String&gt; otherStream = //...

ConnectedStreams&lt;Integer, String&gt; connectedStreams = someStream.connect(otherStream);
 </code></pre> &nbsp; </td> 
 </tr> 
 <tr> 
 <td style="width:204px;">CoMap, CoFlatMap ConnectedStreams → DataStream</td> 
 <td style="width:645px;"> 类似于基于connect数据流上map和flatMap <pre><code>connectedStreams.map(new CoMapFunction&lt;Integer, String, Boolean&gt;() {
 @Override
 public Boolean map1(Integer value) {
 return true;
 }

@Override
    public Boolean map2(String value) {
        return false;
    }
});
connectedStreams.flatMap(new CoFlatMapFunction&lt;Integer, String, String&gt;() {

@Override
   public void flatMap1(Integer value, Collector&lt;String&gt; out) {
       out.collect(value.toString());
   }

@Override
 public void flatMap2(String value, Collector&lt;String&gt; out) {
 for (String word: value.split(" ")) {
 out.collect(word);
 }
 }
});
 </code></pre> &nbsp; </td> 
 </tr> 
 <tr> 
 <td style="width:204px;">Split DataStream → SplitStream</td> 
 <td style="width:645px;"> 根据某些标准将流分成两个或多个流。 <pre><code>SplitStream&lt;Integer&gt; split = someDataStream.split(new OutputSelector&lt;Integer&gt;() {
 @Override
 public Iterable&lt;String&gt; select(Integer value) {
 List&lt;String&gt; output = new ArrayList&lt;String&gt;();
 if (value % 2 == 0) {
 output.add("even");
 }
 else {
 output.add("odd");
 }
 return output;
 }
});
 </code></pre> &nbsp; </td> 
 </tr> 
 <tr> 
 <td style="width:204px;">Select SplitStream → DataStream</td> 
 <td style="width:645px;"> 从拆分流中选择一个或多个流。 <pre><code>SplitStream&lt;Integer&gt; split;
DataStream&lt;Integer&gt; even = split.select("even");
DataStream&lt;Integer&gt; odd = split.select("odd");
DataStream&lt;Integer&gt; all = split.select("even","odd");
 </code></pre> &nbsp; </td> 
 </tr> 
 <tr> 
 <td style="width:204px;">Iterate DataStream → IterativeStream → DataStream</td> 
 <td style="width:645px;"> 通过将一个操作符的输出重定向到前面的某个操作符，在流中创建一个“反馈”循环。这对于定义不断更新模型的算法特别有用。下面的代码从一个流开始，并持续地应用迭代体。大于0的元素被发送回反馈通道，其余的元素被转发到下游。Iterations后续详解//TODO <pre><code>IterativeStream&lt;Long&gt; iteration = initialStream.iterate();
DataStream&lt;Long&gt; iterationBody = iteration.map (/*do something*/);
DataStream&lt;Long&gt; feedback = iterationBody.filter(new FilterFunction&lt;Long&gt;(){
 @Override
 public boolean filter(Long value) throws Exception {
 return value &gt; 0;
 }
});
iteration.closeWith(feedback);
DataStream&lt;Long&gt; output = iterationBody.filter(new FilterFunction&lt;Long&gt;(){
 @Override
 public boolean filter(Long value) throws Exception {
 return value &lt;= 0;
 }
});
 </code></pre> &nbsp; </td> 
 </tr> 
 <tr> 
 <td style="width:204px;">Extract Timestamps DataStream → DataStream</td> 
 <td style="width:645px;">从记录中提取时间戳，以便使用使用事件时间语义的窗口。Event Time后续详解//TODO</td> 
 </tr> 
 <tr> 
 <td style="width:204px;">Project DataStream → DataStream</td> 
 <td style="width:645px;"> 以下转换可用于元组的数据流： 从元组中选择字段的子集 <pre><code>DataStream&lt;Tuple3&lt;Integer, Double, String&gt;&gt; in = // [...]
DataStream&lt;Tuple2&lt;String, Integer&gt;&gt; out = in.project(2,0);</code></pre> &nbsp; </td> 
 </tr> 
 </tbody> 
</table>

### 物理分区 ###

Flink还通过以下功能(如果需要的话)对转换后的流分区进行底层控制。

<table> 
 <tbody> 
 <tr> 
 <td style="width:210px;">方法</td> 
 <td style="width:639px;">描述和代码样例</td> 
 </tr> 
 <tr> 
 <td style="width:210px;">Custom partitioning DataStream → DataStream</td> 
 <td style="width:639px;"> 使用用户定义的分区程序为每个元素选择目标任务。 <pre><code>dataStream.partitionCustom(partitioner, "someKey");
dataStream.partitionCustom(partitioner, 0);
 </code></pre> &nbsp; </td> 
 </tr> 
 <tr> 
 <td style="width:210px;">Random partitioning DataStream → DataStream</td> 
 <td style="width:639px;"> 根据均匀分布随机划分元素。 <pre><code>dataStream.shuffle();
 </code></pre> &nbsp; </td> 
 </tr> 
 <tr> 
 <td style="width:210px;">Rebalancing (Round-robin partitioning) DataStream → DataStream</td> 
 <td style="width:639px;"> 分区元素循环，每个分区创建相同的负载。对于存在数据倾斜的性能优化非常有用。 <pre><code>dataStream.rebalance();</code></pre> &nbsp; </td> 
 </tr> 
 <tr> 
 <td style="width:210px;">Rescaling DataStream → DataStream</td> 
 <td style="width:639px;"> 将元素(循环)划分为下游操作的子集。如果您想要pipelines，例如，从一个源的每个并行实例分散到几个映射器的子集来分配负载，但又不希望rebalance()导致数据的完全重平衡，那么这是非常有用的。这将只需要本地数据传输，而不是通过网络传输数据，这取决于其他配置值，比如TaskManagers的槽数。 上游操作向其发送元素的下游操作子集取决于上游和下游操作的并行度。例如，如果上游操作具有并行性2，下游操作具有并行性6，那么一个上游操作将把元素分配给三个下游操作，而另一个上游操作将分配给另外三个下游操作。另一方面，如果下游操作的并行度为2，上游操作的并行度为6，则上游操作的并行度为3。 在不同并行度不是彼此的倍数的情况下，一个或多个下游操作与上游操作的输入数量不同。 <img alt="" height="200" src="https://img-blog.csdnimg.cn/20190820104338533.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3pwZl85NDA4MTA2NTM4NDI=,size_16,color_FFFFFF,t_70" width="344"> <pre><code>dataStream.rescale();
 </code></pre> &nbsp; </td> 
 </tr> 
 <tr> 
 <td style="width:210px;">Broadcasting DataStream → DataStream</td> 
 <td style="width:639px;"> 向每个分区广播元素。 <pre><code>dataStream.broadcast();</code></pre> &nbsp; </td> 
 </tr> 
 </tbody> 
</table>

### 任务链接和资源组 ###

chaining意味着将它们放在同一个线程中以获得更好的性能。如果可能的话，默认情况下使用Flink的chain操作符(例如，两个后续的映射转换)。如果需要，API可以对链接进行细粒度控制：

如果希望在整个作业中禁用链接，请使用StreamExecutionEnvironment.disableOperatorChaining()。对于更细粒度的控制，可以使用以下函数。注意，这些函数只能在DataStream转换之后使用，因为它们引用的是前面的转换。例如，可以使用someStream.map(…). startnewchain()，但是不能使用someStream.startNewChain()。

resource group是Flink中的一个插槽，请参见插槽//TODO。如果需要，可以手动将operators隔离在单独的槽中。

<table> 
 <tbody> 
 <tr> 
 <td style="width:214px;">方法</td> 
 <td style="width:635px;">描述和代码样例</td> 
 </tr> 
 <tr> 
 <td style="width:214px;">Start new chain</td> 
 <td style="width:635px;"> 从这个操作符开始，创建一个新的chain。这两个映射器将被链接，过滤器将不会链接到第一个映射器。 <pre><code>someStream.filter(...).map(...).startNewChain().map(...);</code></pre> &nbsp; </td> 
 </tr> 
 <tr> 
 <td style="width:214px;">Disable chaining</td> 
 <td style="width:635px;"> <pre><code>Do not chain the map operator </code></pre> &nbsp; </td> 
 </tr> 
 <tr> 
 <td style="width:214px;"> Do not chain the map operator </td> 
 <td style="width:635px;"> 设置操作的槽共享组。Flink将把具有相同槽共享组的operator放入相同槽中，同时将没有槽共享组的operator保留在其他槽中。这可以用来隔离插槽。如果所有输入operator都位于同一个槽共享组中，则槽共享组将从输入operator继承。默认槽共享组的名称为“default”，可以通过调用slotSharingGroup(“default”)显式地将操作放入该组。 <pre><code>someStream.filter(...).slotSharingGroup("name");</code></pre> &nbsp; </td> 
 </tr> 
 </tbody> 
</table>