java 大数据 31、Flink 的 DataStream API 数据流算子详解

1.算子

可以通过算子将一个或多个 DataStream 转换成新的 DataStream，也可以将多个数据转换算子合并成一个复杂的数据流拓扑。

2.数据流转换

a）Map

DataStream → DataStream

输入一个元素，转换后输出一个元素，示例将输入流中元素数值加倍。

DataStream dataStream = //...

dataStream.map(new MapFunction() {

@Override

public Integer map(Integer value) throws Exception {

return 2 * value;

}

});

b）FlatMap

DataStream → DataStream

输入一个元素，转换后产生零个、一个或多个元素，示例将句子拆分为单词。

dataStream.flatMap(new FlatMapFunction() {

@Override

public void flatMap(String value, Collector out)

throws Exception {

for(String word: value.split(" ")){

out.collect(word);

}

}

});

c）Filter

DataStream → DataStream

为每个元素执行一个布尔 function，并保留那些 function 输出值为 true 的元素，示例过滤掉零值。

dataStream.filter(new FilterFunction() {

@Override

public boolean filter(Integer value) throws Exception {

return value != 0;

}

});

d）KeyBy

DataStream → KeyedStream

在逻辑上将流划分为不相交的分区，具有相同 key 的记录都分配到同一个分区；在内部， keyBy() 是通过哈希分区实现的。

dataStream.keyBy(value -> value.getSomeKey());

dataStream.keyBy(value -> value.f0);

以下情况，一个类不能作为 key：

它是一种 POJO 类，但没有重写 hashCode() 方法而是依赖于 Object.hashCode() 实现。它是任意类的数组。

e）Reduce

KeyedStream → DataStream

在相同 key 的数据流上“滚动”执行 reduce，将当前元素与最后一次 reduce 得到的值组合然后输出新值，示例局部求和。

keyedStream.reduce(new ReduceFunction() {

@Override

public Integer reduce(Integer value1, Integer value2)

throws Exception {

return value1 + value2;

}

});

f）Window

KeyedStream → WindowedStream

在已经分区的 KeyedStreams 上定义 Window，Window 根据某些特征（例如，最近 5 秒内到达的数据）对每个 key Stream 中的数据进行分组。

dataStream

.keyBy(value -> value.f0)

.window(TumblingEventTimeWindows.of(Time.seconds(5)));

g）WindowAll

DataStream → AllWindowedStream

在 DataStream 上定义 Window，Window 根据某些特征（例如，最近 5 秒内到达的数据）对所有流事件进行分组。

注意：适用于非并行转换的大多数场景，所有记录都将收集到 windowAll 算子对应的一个任务中。

dataStream

.windowAll(TumblingEventTimeWindows.of(Time.seconds(5)));

h）Window Apply

WindowedStream → DataStream

AllWindowedStream → DataStream

将 function 应用于整个窗口，示例手动对窗口内元素求和。

windowedStream.apply(new WindowFunction, Integer, Tuple, Window>() {

public void apply (Tuple tuple,

Window window,

Iterable> values,

Collector out) throws Exception {

int sum = 0;

for (value t: values) {

sum += t.f1;

}

out.collect (new Integer(sum));

}

});

// 在 non-keyed 窗口流上应用 AllWindowFunction

allWindowedStream.apply (new AllWindowFunction, Integer, Window>() {

public void apply (Window window,

Iterable> values,

Collector out) throws Exception {

int sum = 0;

for (value t: values) {

sum += t.f1;

}

out.collect (new Integer(sum));

}

});

注意：如果使用 windowAll 转换，则需要改用 AllWindowFunction。

i）WindowReduce

WindowedStream → DataStream

对窗口应用 reduce function 并返回 reduce 后的值。

windowedStream.reduce (new ReduceFunction>() {

public Tuple2 reduce(Tuple2 value1, Tuple2 value2) throws Exception {

return new Tuple2(value1.f0, value1.f1 + value2.f1);

}

});

j）Union

DataStream → DataStream*

将两个或多个数据流联合来创建一个包含所有流中数据的新流。

注意：如果一个数据流和自身进行联合，这个流中的每个数据将在合并后的流中出现两次。

dataStream.union(otherStream1, otherStream2, ...);

k）Window Join

DataStream,DataStream → DataStream

根据指定的 key 和窗口 join 两个数据流。

dataStream.join(otherStream)

.where().equalTo()

.window(TumblingEventTimeWindows.of(Time.seconds(3)))

.apply (new JoinFunction () {...});

l）Interval Join

KeyedStream,KeyedStream → DataStream

根据 key 相等并且在指定的时间范围内（e1.timestamp + lowerBound <= e2.timestamp <= e1.timestamp + upperBound）的条件将分别属于两个 keyed stream 的元素 e1 和 e2 Join 在一起。

// this will join the two streams so that

// key1 == key2 && leftTs - 2 < rightTs < 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() {...});

m）Window CoGroup

DataStream,DataStream → DataStream

根据指定的 key 和窗口将两个数据流组合在一起。

dataStream.coGroup(otherStream)

.where(0).equalTo(1)

.window(TumblingEventTimeWindows.of(Time.seconds(3)))

.apply (new CoGroupFunction () {...});

n）Connect

DataStream,DataStream → ConnectedStream

“连接” 两个数据流并保留各自的类型，connect 允许在两个流的处理逻辑之间共享状态。

DataStream someStream = //...

DataStream otherStream = //...

ConnectedStreams connectedStreams = someStream.connect(otherStream);

o）CoMap, CoFlatMap

ConnectedStream → DataStream

在连接的数据流上进行 map 和 flatMap。

connectedStreams.map(new CoMapFunction() {

@Override

public Boolean map1(Integer value) {

return true;

}

@Override

public Boolean map2(String value) {

return false;

}

});

connectedStreams.flatMap(new CoFlatMapFunction() {

@Override

public void flatMap1(Integer value, Collector out) {

out.collect(value.toString());

}

@Override

public void flatMap2(String value, Collector out) {

for (String word: value.split(" ")) {

out.collect(word);

}

}

});

p）Cache

DataStream → CachedDataStream

把算子的结果缓存起来，目前只支持批执行模式下运行的作业。

算子的结果在算子第一次执行的时候会被缓存起来，之后的 作业中会复用该算子缓存的结果；如果算子的结果丢失了，它会被原来的算子重新计算并缓存。

DataStream dataStream = //...

CachedDataStream cachedDataStream = dataStream.cache();

cachedDataStream.print(); // Do anything with the cachedDataStream

...

env.execute(); // Execute and create cache.

cachedDataStream.print(); // Consume cached result.

env.execute();

参考文章