美文网首页数客联盟
Flink 异步IO实战

Flink 异步IO实战

作者: Woople | 来源:发表于2020-01-01 05:01 被阅读0次

    基本概念

    首先通过官网的一个图片了解一下Asynchronous I/O Operation

    Flink source收到一条数据就会进行处理,如果需要通过这条数据关联外部数据源,例如mysql,在发出查询请求后,同步IO的方式是会等待查询结果再处理下一条数据的查询,也就是每一条数据都要等待上一个查询结束。而异步IO是指数据来了以后发出查询请求,先不等查询结果,直接继续发送下一条的查询请求,对于查询结果是异步返回的,返回结果之后再进入下一个算子的计算。这两种方式性能差距请看下的样例。

    样例

    代码传送门
    生成6条数据,从0开始递增的6个数字。模拟异步查询之后,加上时间戳输出

    public class AsyncIODemo {
        public static void main(String[] args) throws Exception {
            StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
            env.setParallelism(1);
            final int maxCount = 6;
            final int taskNum = 1;
            final long timeout = 40000;
    
            DataStream<Integer> inputStream = env.addSource(new SimpleSource(maxCount));
            AsyncFunction<Integer, String> function = new SampleAsyncFunction();
    
            DataStream<String> result = AsyncDataStream.unorderedWait(
                        inputStream,
                        function,
                        timeout,
                        TimeUnit.MILLISECONDS,
                        10).setParallelism(taskNum);
    
            result.map(new MapFunction<String, String>() {
                @Override
                public String map(String value) throws Exception {
                    return value + "," + System.currentTimeMillis();
                }
            }).print();
    
            env.execute("Async IO Demo");
        }
    
        private static class SimpleSource implements SourceFunction<Integer> {
            private volatile boolean isRunning = true;
            private int counter = 0;
            private int start = 0;
    
            public SimpleSource(int maxNum) {
                this.counter = maxNum;
            }
    
            @Override
            public void run(SourceContext<Integer> ctx) throws Exception {
                while ((start < counter || counter == -1) && isRunning) {
                    synchronized (ctx.getCheckpointLock()) {
                        System.out.println("send data:" + start);
                        ctx.collect(start);
                        ++start;
                    }
                    Thread.sleep(10L);
                }
            }
    
            @Override
            public void cancel() {
                isRunning = false;
            }
        }
    }
    

    异步方法

    public class SampleAsyncFunction extends RichAsyncFunction<Integer, String> {
        private long[] sleep = {100L, 1000L, 5000L, 2000L, 6000L, 100L};
    
        @Override
        public void open(Configuration parameters) throws Exception {
            super.open(parameters);
        }
    
        @Override
        public void close() throws Exception {
            super.close();
        }
    
        @Override
        public void asyncInvoke(final Integer input, final ResultFuture<String> resultFuture) {
            System.out.println(System.currentTimeMillis() + "-input:" + input + " will sleep " + sleep[input] + " ms");
    
            query(input, resultFuture);
        }
    
        private void query(final Integer input, final ResultFuture<String> resultFuture) {
            try {
                Thread.sleep(sleep[input]);
                resultFuture.complete(Collections.singletonList(String.valueOf(input)));
            } catch (InterruptedException e) {
                resultFuture.complete(new ArrayList<>(0));
            }
        }
    
        private void asyncQuery(final Integer input, final ResultFuture<String> resultFuture) {
            CompletableFuture.supplyAsync(new Supplier<Integer>() {
    
                @Override
                public Integer get() {
                    try {
                        Thread.sleep(sleep[input]);
                        return input;
                    } catch (Exception e) {
                        return null;
                    }
                }
            }).thenAccept((Integer dbResult) -> {
                resultFuture.complete(Collections.singleton(String.valueOf(dbResult)));
            });
        }
    }
    

    上面的代码中有两个方法query()asyncQuery(),其中Thread.sleep(sleep[input]);用来模拟查询需要等待的时间,每条数据等待的时间分别为100L, 1000L, 5000L, 2000L, 6000L, 100L毫秒。

    结果分析

    运行query()的结果为

    send data:0
    send data:1
    send data:2
    send data:3
    send data:4
    send data:5
    1577801193230-input:0 will sleep 100 ms
    1577801193331-input:1 will sleep 1000 ms
    0,1577801194336
    1,1577801194336
    1577801194336-input:2 will sleep 5000 ms
    1577801199339-input:3 will sleep 2000 ms
    2,1577801201341
    1577801201342-input:4 will sleep 6000 ms
    3,1577801207345
    4,1577801207345
    1577801207346-input:5 will sleep 100 ms
    5,1577801207451
    

    可以看到第一条数据进入到map算子的时间与最后一条相差了13115毫秒,执行的顺序与source中数据的顺序一致,并且是串行的。

    运行asyncQuery()的结果为

    send data:0
    send data:1
    send data:2
    send data:3
    1577802161755-input:0 will sleep 100 ms
    1577802161756-input:1 will sleep 1000 ms
    1577802161757-input:2 will sleep 5000 ms
    send data:4
    send data:5
    1577802161783-input:3 will sleep 2000 ms
    1577802161784-input:4 will sleep 6000 ms
    1577802161785-input:5 will sleep 100 ms
    0,1577802161859
    1,1577802162759
    3,1577802163862
    5,1577802163962
    2,1577802166760
    4,1577802168762
    

    同样第一条数据进入map算子的时间与最后一条仅相差了6903毫秒,而且输出结果的顺序并不是source中的顺序,而是按照查询时间递增的顺序输出,并且查询请求几乎是同一时间发出的。

    通过上面的例子可以看出,flink所谓的异步IO,并不是只要实现了asyncInvoke方法就是异步了,这个方法并不是异步的,而是要依靠这个方法里面所写的查询是异步的才可以。否则像是上面query()方法那样,同样会阻塞查询相当于同步IO。在实现flink异步IO的时候一定要注意。官方文档也给出了相关的说明。

    For example, the following patterns result in a blocking asyncInvoke(...) functions and thus void the asynchronous behavior:Using a database client whose lookup/query method call blocks until the result has been received back

    总结

    本文基于flink 1.9.0。通过样例介绍了如何实现flink异步IO,读者可以修改本文样例体验异步IO其他的特性,例如Order of Results或者Event Time

    相关文章

      网友评论

        本文标题:Flink 异步IO实战

        本文链接:https://www.haomeiwen.com/subject/qdvioctx.html