flink key通过增加延迟；如何减少此延迟? [英] flink keyBy adding delay; how can I reduce this latency?

问题描述

当我使用KeyedStream运行一个简单的flink应用程序时，我观察到事件的时间延迟在0到100毫秒之间变化.下面是程序

When I ran a simple flink application with KeyedStream, I observed the time latency of an event varies from 0 to 100ms. Below is the program

StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
    DataStream<Long> source = env.addSource(new SourceFunction<Long>() {
        public void run(SourceContext<Long> sourceContext) throws Exception {
            while(true) {
                synchronized (sourceContext.getCheckpointLock()) {
                    sourceContext.collect(System.currentTimeMillis());
                    Thread.sleep(1000);
                }
            }
        }

        public void cancel() {}
    }).keyBy(new KeySelector<Long, Long>() {
        @Override
        public Long getKey(Long l) throws Exception {
            return l;
        }
    }).addSink(new SinkFunction<Long>() {
        @Override
        public void invoke(Long l) throws Exception {
            long diff = System.currentTimeMillis() - l;
            System.out.println("in Sink: diff=" + diff);
        }
    });
    env.execute();

输出为:

in Sink: diff=0
in Sink: diff=2
in Sink: diff=4
in Sink: diff=4
in Sink: diff=5
in Sink: diff=7
in Sink: diff=9
in Sink: diff=9
in Sink: diff=11
in Sink: diff=12
in Sink: diff=14
in Sink: diff=14
in Sink: diff=16
in Sink: diff=17
in Sink: diff=18
in Sink: diff=19
in Sink: diff=21
in Sink: diff=22
in Sink: diff=24
in Sink: diff=24
in Sink: diff=26
in Sink: diff=27
in Sink: diff=29
in Sink: diff=29
in Sink: diff=31
in Sink: diff=32
in Sink: diff=34
in Sink: diff=34
in Sink: diff=36
in Sink: diff=37
in Sink: diff=39
in Sink: diff=40
in Sink: diff=41
in Sink: diff=43
in Sink: diff=45
in Sink: diff=45
in Sink: diff=47
in Sink: diff=48
in Sink: diff=50
in Sink: diff=50
in Sink: diff=52
in Sink: diff=53
in Sink: diff=55
in Sink: diff=57
in Sink: diff=57
in Sink: diff=59
in Sink: diff=60
in Sink: diff=61
in Sink: diff=62
in Sink: diff=63
in Sink: diff=65
in Sink: diff=66
in Sink: diff=67
in Sink: diff=69
in Sink: diff=70
in Sink: diff=72
in Sink: diff=72
in Sink: diff=74
in Sink: diff=76
in Sink: diff=77
in Sink: diff=78
in Sink: diff=79
in Sink: diff=81
in Sink: diff=82
in Sink: diff=83
in Sink: diff=84
in Sink: diff=86
in Sink: diff=87
in Sink: diff=88
in Sink: diff=89
in Sink: diff=91
in Sink: diff=92
in Sink: diff=94
in Sink: diff=94
in Sink: diff=96
in Sink: diff=97
in Sink: diff=99
in Sink: diff=99
in Sink: diff=0
in Sink: diff=2
in Sink: diff=3
in Sink: diff=4
in Sink: diff=4
in Sink: diff=5
in Sink: diff=7
in Sink: diff=9
in Sink: diff=9
in Sink: diff=11
in Sink: diff=12
in Sink: diff=14
in Sink: diff=14
in Sink: diff=16
in Sink: diff=17
in Sink: diff=18
in Sink: diff=19
in Sink: diff=21
in Sink: diff=22
in Sink: diff=24
in Sink: diff=24
in Sink: diff=26
in Sink: diff=46
in Sink: diff=48
in Sink: diff=50
in Sink: diff=52
in Sink: diff=53
in Sink: diff=54
in Sink: diff=56
in Sink: diff=58
in Sink: diff=59
in Sink: diff=60
in Sink: diff=62
in Sink: diff=64
in Sink: diff=65
in Sink: diff=66
in Sink: diff=68
in Sink: diff=70
in Sink: diff=71
in Sink: diff=73
in Sink: diff=74
in Sink: diff=76
in Sink: diff=77
in Sink: diff=79
in Sink: diff=81
in Sink: diff=82
in Sink: diff=83
in Sink: diff=85
in Sink: diff=86
in Sink: diff=88
in Sink: diff=88
in Sink: diff=90
in Sink: diff=92
in Sink: diff=92
in Sink: diff=94
in Sink: diff=95
in Sink: diff=97
in Sink: diff=98
in Sink: diff=99
in Sink: diff=0
in Sink: diff=2
in Sink: diff=4
in Sink: diff=4
in Sink: diff=5
in Sink: diff=7
in Sink: diff=9

如您所见，等待时间是一个模式逐渐增加到100，然后下降并从0开始，并且循环重复.我需要等待时间尽可能短.此示例是我的实际应用程序的简化版本.有人可以向我解释延迟的原因以及如何将延迟降低到尽可能低的水平.

As you can see the latency is a pattern gradually increases to 100 and the drops and starts from 0 and the cycle repeats. I need the latency to be as low as possible. This example is a simplified version of my real application. Can someone explain me the reason for latency and how to reduce it to as low as possible.

推荐答案

此延迟的原因是，通过添加keyBy，您将迫使网络改组以及序列化/反序列化.延迟如此可变的原因是由于网络缓冲.

The reason for this delay is that by adding that keyBy you are forcing a network shuffle along with serialization/deserialization. The reason the delay is so variable is because of the network buffering.

您将要阅读文档中名为控制延迟. tl; dr是您要将网络缓冲区超时设置为较小值:

You'll want to read the section of the documentation called Controlling Latency. The tl;dr is that you want to set the network buffer timeout to something small:

env.setBufferTimeout(timeoutMillis);

如果需要，您可以将缓冲区超时设置为零，但是与将其设置为较小的值(例如1ms或5ms)相比，它对吞吐量的影响更大.默认值为100毫秒.有关如何组织Flink中的网络堆栈的详细信息，请参见 A在Flink项目博客上深入研究Flink的网络堆栈.

You can set the buffer timeout to zero if you want, but that will impact throughput more than setting it to something small (like 1ms, or 5ms). The default is 100ms. For details on how the network stack in Flink is organized, see A Deep-Dive into Flink's Network Stack on the Flink project blog.

在我们讨论这个问题的同时，其他延迟源可能包括检查点障碍对齐和垃圾回收.

While we're on the subject, other sources of latency can include checkpoint barrier alignment and garbage collection.

env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.AT_LEAST_ONCE);

将禁用障碍对齐，但以完全放弃一次处理语义为代价.

will disable barrier alignment, at the cost of giving up exactly once processing semantics.

使用RocksDB状态后端将减少要进行垃圾回收的对象数量(因为RocksDB使其状态处于堆外状态)，在某些情况下，以最差的平均延迟为代价提高了最坏的延迟.但是，对于使用RocksDB来改善最坏情况的延迟的现代垃圾收集器来说，这可能是一个错误.

Using the RocksDB state backend will reduce the number of objects to garbage collect (since RocksDB keeps its state off-heap), in some cases improving worst case latency at the cost of worse average latency. However, with modern garbage collectors using RocksDB to improve worst-case latency can be a mistake.

还

env.getConfig().enableObjectReuse();

将指示运行时重用用户对象以获得更好的性能.请记住，当用户代码功能不了解此行为时，这可能会导致错误.

will instruct the runtime to reuse user objects for better performance. Keep in mind that this can lead to bugs when user-code functions are not aware of this behavior.

如果使用水印，则水印延迟会影响触发事件时间计时器(包括Windows)和

If you are using watermarks, the watermark delay affects the latency with which event time timers will be triggered (including windows), and the autoWatermarkInterval also has an impact on latency.

最后，使用事务接收器会增加端到端延迟，因为这些接收器的下游使用者在事务完成之前不会看到提交的结果.预期的延迟大约是检查点间隔的一半.

Finally, the use of transactional sinks adds end-to-end latency, since downstream consumers of those sinks won't see committed results until the transactions complete. The expected delay is roughly half the checkpoint interval.

如果您对测量延迟感兴趣，请查看延迟跟踪和监视Apache Flink应用程序101 .

If you are interested in measuring latency, take a look at Latency Tracking and the section on latency in Monitoring Apache Flink Applications 101.

这篇关于flink key通过增加延迟；如何减少此延迟?的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！