我的拓扑结构如下： kafka(p:6)->reduce(p:6)->db writer(p:12) （其中p：平行度）。
我让它在一个节点“集群”上运行 taskmanager.numberOfTaskSlots: 30 我知道我的Kafka音源每分钟产生650万张唱片
Kafka“读取器”的并行度等于Kafka分区的#
当我观察这份工作（通过flink ui）约1分钟时，这些是我看到的值：
Kafka->减少：发送约150万条记录（减少>4倍）
reduce（加窗聚合5秒）->db write发送约114k条记录（off by>2x）1
接收到的db write-->记录：~23k（关断>5x）2
（其他零件的发送/接收值之间的差异较小，但我可以将其归因于测量误差）
问题：
1剩下的记录呢？
2这台机器运行时，负载永远不会超过1.5。还有其他限制因素吗？
三。我是否误读了ui中的值？
java 8
flink 1.0（最新github）
机器：32核/96 gb ram
1这可以用聚合过程来解释。
2该值与写入数据库的内容一致。
1条答案
按热度按时间j8ag8udp1#
Flink没有丢失记录，他们只是在飞行中缓冲，或者他们在Kafka停留的时间更长。从数字上看，你似乎正在经历背压。
您可以看到，“reducer”发出了许多“db writer”尚未接收到的记录。在这种情况下，这些记录仍在操作员之间通信通道的缓冲区中。这些通道的缓冲量有限（取决于配置的缓冲区的数量，通常为几MB）。对于小的记录，他们可能会持有一些多个10公里的记录。
如果一个运营商中发送的记录数始终明显落后于接收运营商中接收的记录数，则这表明接收器（此处为“db writer”）无法跟上数据速率。可能是因为db处理插入的速度不够快（太同步，太细粒度提交？），可能是“db writer”和db之间的网络饱和了。
在这种情况下，“db writer”将反压减压器，这最终也将反压Kafka的来源。
为了测试如果没有来自数据库的背压，数据速率会是多少，可以尝试一个“db writer”简单地删除所有记录的实验。