flink：处理背压（来源：kafka，sink：elasticsearch）

Question

我有一份flink的工作，从Kafka读取数据，执行某些聚合，并将结果写入elasticsearch索引。我看到震源上有很高的背压。高背压导致数据从Kafka缓慢读取，我看到数据在网络堆栈中排队（netstat RecvQ显示源Kafka连接中有上万字节的数据，数据最终被读取），这反过来会导致数据在延迟后沉入elasticsearch，并且延迟持续增加。

源每分钟产生约17500条记录，Flink作业为每个传入记录分配（事件）时间戳，执行12种不同类型的keyBy，在1分钟的翻滚窗口中匹配事件，在此键控的windows流上执行聚合操作，最后将结果写入12个不同的elasticsearch索引（每次写入都是插入）。

问题是写入elasticsearch的数据开始滞后，因此仪表板结果（建立在elasticsearch之上）不再是实时的。我的理解是，这是因为背压增加而发生的。不确定如何解决这个问题。html" target="_blank">集群本身是一个基于VM的单节点独立集群，具有64GB RAM（任务管理器配置为使用20GB）和16个vCPU。没有证据（从htop中看到）CPU或内存受到限制。只有一个任务管理器，这是此集群上唯一的flink作业。

我不确定这个问题是由于集群中的一些本地资源问题还是由于写入elasticsearch速度慢。我已将setBulkFlushMaxActions设置为1（正如我在任何地方看到的所有代码示例中所做的那样），我是否需要设置setBulkFlushInterval和/或setBulkFlushMaxSizeinMB？

我经历过https://www.da-platform.com/flink-forward-berlin/resources/improving-throughput-and-latency-with-flinks-network-stack但尚未尝试幻灯片19中列出的调整选项，不确定要为这些参数设置什么值。

最后，我认为在IntelliJ IDE中运行同一个作业时，我看不到同样的问题。

我将排除所有聚合，然后逐个将它们添加回来，看看是否有特定的聚合触发此问题？

任何特定的指针将非常感激，也将尝试setBulkFlushInterval和setBulkFlushMaxSizeinMB。

2019年1月29日更新1似乎两个节点都在以非常高的堆使用率运行，因此GC一直在运行，试图清除JVM中的空间。将物理内存从16 GB增加到32GB，然后重新启动节点。这将有望解决问题，24小时后就会知道。

Answer 1

通常在这种情况下，问题在于与外部数据存储的连接——要么带宽不足，要么对每条记录进行同步写入，而不是成批写入。

验证elasticsearch接收器是否是问题（而不是网络堆栈配置）的一种简单方法是将其替换为丢弃接收器（一个什么都不做的接收器），看看这是否解决了问题。类似的东西

public static class NullSink<OUT> implements SinkFunction<OUT> {
    @Override
    public void invoke(OUT value, Context context) throws Exception {
    }
}

更新：

问题是您设置了批量。脸红将max.actions设置为1，防止连接到elasticsearch服务器时出现任何缓冲。

Answer 2

通过增加（加倍）elasticsearch集群节点上的RAM并将索引刷新间隔（在所有elasticsearch索引上）设置为30（默认为1秒），问题得到了解决。进行这些更改后，闪烁中的背压报告为ok，没有数据延迟，一切看起来都很好。