java - 弗林克 : is it losing records?

Question

我的拓扑是这样的:kafka(p:6)->reduce(p:6)->db writer(p:12)(其中p:是并行)。

• 我让它在单个节点“集群”上运行 taskmanager.numberOfTaskSlots: 30
• 我知道我的 kafka 源每分钟产生约 650 万条记录
• kafka 'reader' 的并行度等于 kafka 分区的数量

当我(通过 flink UI)观察这项工作约 1 分钟时，这些是我看到的值:

• kafka -> 减少:发送了约 150 万条记录(减少了 > 4 倍)
• 减少(窗口聚合 5 秒)-> 数据库写入 ~114K 条记录已发送(减少 > 2 倍)¹
• db write --> records received: ~23K (off by > 5x) ²

(其他部分的发送/接收值之间存在较小的差异，但我可以将这些归因于测量误差)

问题:
1. 剩下的记录在哪里？
2. 运行时这台机器上的负载永远不会超过 1.5。还有其他限制因素吗？
3. 我是否误读了 UI 中的值？

Java 8
Flink 1.0(最新github)
机器:32 核/96 Gb RAM

¹这个可以用聚合过程来解释。
²这个值与写入数据库的内容一致。

Answer 1

Flink 不会丢失记录，它们只是在飞行中被缓冲，或者在 Kafka 中停留更长时间。从数字来看，您似乎正在经历背压。

您可以看到“reducer”发出了很多尚未被“db writer”接收的记录。在那种情况下，这些记录仍在运算符(operator)之间的通信 channel 的缓冲区中。这些 channel 的缓冲量有限(取决于配置的缓冲区数量，通常为几 MB)。对于小记录，它们可能会保存多个 10k 记录。

如果一个运算符(operator)发送的记录数持续明显落后于接收运算符(operator)接收的记录数，这表明接收器(此处为“数据库写入器”)无法跟上数据速率。可能是因为数据库处理插入的速度不够快(太同步，太细粒度提交？)，也许“数据库编写器”和数据库之间的网络已经饱和。

在这种情况下，“db writer”将对 reducer 进行反压，最终也会对 Kafka Source 进行反压。

如果您没有来自数据库的背压，要尝试数据速率是多少，您可以尝试一个“数据库编写器”简单地删除所有记录的实验。