将KafkaFlink到S3并重试

Question

我是Flink的新手，我需要从Kafka读取数据，通过使用一些API有条件地丰富这些数据（如果记录属于X类），然后写入S3。

我用上面的逻辑制作了一个你好世界Flink应用程序，它就像一个魅力。

但是，我用来充实的API没有100%的正常运行时间SLA，所以我需要设计一些带有重试逻辑的东西。

以下是我找到的选项，

选项1）进行指数重试，直到从API获得响应，但这会阻塞队列，所以我不喜欢这样

选项2）使用另一个主题（称为主题失败），并在API关闭时将其发布到主题失败。这样就不会阻塞实际的主队列。我还需要一个工人来处理队列主题failure中的数据。同样，如果API长时间停机，则必须将此队列用作循环队列。例如，从队列主题失败中读取消息，如果无法将其推送到名为主题失败的同一队列并使用队列主题失败中的下一条消息，请尝试充实。

我更喜欢选项2，但这看起来不是一件容易的事情。是否有任何标准的Flink方法可用于实现选项2？

Answer 1

这是从微服务迁移时出现的一个相当常见的问题。正确的解决方案是将查找数据也放在Kafka或某些DB中，这些DB可以作为附加源集成到同一Flink应用程序中。

如果您无法做到这一点（例如，API是外部的或数据无法轻松映射到数据存储），这两种方法都是可行的，并且它们具有不同的优势。

1） 将允许您保留输入事件的顺序。如果下游应用程序需要有序性，则需要重试。

2） 常见的术语是死信队列（尽管更常用于无效记录）。在Flink中有两种简单的集成方法，要么有一个单独的源，要么使用一个主题模式/列表和一个源。

您的拓扑如下所示：

Kafka Source      -\             Async IO        /-> Filter good -> S3 sink
                    +-> Union -> with timeout  -+ 
Kafka Source dead -/           (for API call!)   \-> Filter bad  -> Kafka sink dead