elasticsearch - flink : handling backpressure (source: kafka, 接收器:elasticsearch)

Question

我有一个 flink 作业，它从 Kafka 读取数据，执行某些聚合并将结果写入 elasticsearch 索引。我在源上看到高背压。高背压导致数据从 Kafka 读取缓慢，我看到数据在网络堆栈中排队(netstat RecvQ 显示数万字节的数据卡在源 kafka 连接中，数据最终被读取)这反过来导致数据在滞后后下沉到elasticsearch，并且这个滞后不断增加。

源每分钟生成约 17,500 条记录，Flink 作业为每个传入记录分配(事件)时间戳，执行 12 种不同类型的 keyBy，将事件放入 1 分钟的滚动窗口中，对该键控窗口流执行聚合操作最后将结果写入12个不同的elasticsearch索引(每次写入都是一次插入)。

问题是写入 elasticsearch 的数据开始滞后，因此仪表板结果(建立在 elasticsearch 之上)不再是实时的。我的理解是，这是由于背压增加而发生的。不知道如何解决这个问题。集群本身是基于 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。

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

Answer 1

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