flink使用08-在dataStream中使用AsyncFunction

在流式处理的过程中, 在中间步骤的处理中, 如果涉及到一些费事的操作或者是外部系统的数据交互, 那么就会给整个流造成一定的延迟. 在 flink 的 1.2 版本中引入了 Asynchronous I/O, 能够支持异步的操作, 以提高 flink 系统与外部数据系统交互的性能及吞吐量.

在使用 Flink 的异步 IO 时, 主要有两个 API可以使用, 一个是AsyncDataStream.unorderedWait( ), 另一个AsyncDataStream.orderedWait( ).在异步处理过程中,原本数据的顺序可能会发生变化, 使用unorderWait的方法, 不会考虑顺序的问题, 一旦处理完成就会直接返回结果, 这种方法具有较低的延迟和负载. 那么orderWait的方法就是想对应的, 严格按照原本流中的数据顺序做返回, 会对系统造成一定的延迟. 实际中应该根据具体的业务情况做选择.unorderedWait或orderedWait有两个关于async operation的参数,一个是timeout参数用于设置async的超时时间,一个是capacity参数用于指定同一时刻最大允许多少个(并发)async request在执行;

在使用异步IO时,需要自己去继承AsyncFunction,AsyncFunction接口继承了Function,它定义了asyncInvoke方法以及一个default的timeout方法;asyncInvoke方法执行异步逻辑,然后通过ResultFuture.complete将结果或异常设置到ResultFuture,如果异常则通过ResultFuture.completeExceptionally(Throwable)来传递 ResultFuture;RichAsyncFunction继承了AbstractRichFunction,同时声明实现AsyncFunction接口,它不没有实现asyncInvoke,交由子类实现;它覆盖了setRuntimeContext方法,这里使用RichAsyncFunctionRuntimeContext或者RichAsyncFunctionIterationRuntimeContext进行包装.

下面是一个验证 Async I/O 的demo, 具体代码见仓库 -> code link

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
public class AsyncIOExample {
public static void main(String[] args) throws Exception {

final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

DataStream<String> inp = env.fromElements(AsyncIOData.WORDS);
// 接收数据
SingleOutputStreamOperator<String> out = inp.map(new MapFunction<String, String>() {
@Override
public String map(String s) throws Exception {
System.out.println("读取数据:" + s + " 当前时间:" + System.currentTimeMillis());
return s;
}
});
// 使用 AsyncFunction 对函数做一个简单的处理, 中间随机睡眠 1-10s
DataStream<String> asyncStream = AsyncDataStream.unorderedWait(out, new SimpleAsyncFunction(), 20_000L, TimeUnit.MILLISECONDS);
// 对已经被 AsyncFunction 处理过的数据再输出一次
asyncStream.map(new MapFunction<String, String>() {
@Override
public String map(String s) throws Exception {
System.out.println("数据处理完毕:" + s + " 当前时间:" + System.currentTimeMillis());
return s;
}
});


env.execute("AsyncFunction Demo");
}

public static class SimpleAsyncFunction extends RichAsyncFunction<String, String>{

private long waitTime;
private final Random rnd = new Random(hashCode());

@Override
public void asyncInvoke(String input, ResultFuture<String> resultFuture) throws Exception {
// 随机睡眠 1 - 10s
System.out.println("开始 AsyncFunction target -> " + input);
waitTime = rnd.nextInt(10);
Thread.sleep(waitTime * 1000);
String out = input + input;
resultFuture.complete(Collections.singletonList(out));
System.out.println("结束 AsyncFunction target -> " + input + " Sleep time = " + waitTime + "s");
}
}
}

以上代码的输出结果为:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
读取数据:D  当前时间:1569574233046
读取数据:C 当前时间:1569574233047
读取数据:A 当前时间:1569574233048
读取数据:B 当前时间:1569574233049
开始 AsyncFunction target -> D
开始 AsyncFunction target -> C
开始 AsyncFunction target -> A
开始 AsyncFunction target -> B
结束 AsyncFunction target -> DSleep time = 6s
数据处理完毕:DD 当前时间:1569574239065
结束 AsyncFunction target -> CSleep time = 6s
数据处理完毕:CC 当前时间:1569574239069
结束 AsyncFunction target -> ASleep time = 6s
数据处理完毕:AA 当前时间:1569574239072
结束 AsyncFunction target -> BSleep time = 6s
数据处理完毕:BB 当前时间:1569574239076
0%