基于流的应用程序中的受控/手动错误/恢复处理

Question

我正在开发一个基于Apache Flink的应用程序，它使用Apache Kafka进行输入和输出。可能这个应用程序将被移植到Apache Spark，所以我也将其添加为标签，问题仍然相同。

我要求必须按顺序处理通过kafka接收的所有传入消息，并且安全地存储在持久层（数据库）中，并且不得丢失任何消息。

这个应用程序中的流媒体部分相当简单/小，因为主要逻辑将归结为类似：

environment.addSource(consumer)    // 1) DataStream[Option[Elem]]
  .filter(_.isDefined)             // 2) discard unparsable messages
  .map(_.get)                      // 3) unwrap Option
  .map(InputEvent.fromXml(_))      // 4) convert from XML to internal representation
  .keyBy(_.id)                     // 5) assure in-order processing on logical-key level
  .map(new DBFunction)             // 6) database lookup, store of update and additional enrichment
  .map(InputEvent.toXml(_))        // 7) convert back to XML
  .addSink(producer)               // 8) attach kafka producer sink

现在，在此管道中，可能会出现几种错误情况：

• 数据库变得不可用（关闭，表空间已满，......）
• 由于逻辑错误（来自列格式）无法存储更改
• 由于代理不可用，kafka生产者无法发送消息

可能还有其他情况。

现在我的问题是，在这些情况下，当我实际上必须做以下事情时，如何确保按照上述方式保持一致性：

1. Stream-Operator 6）检测到问题（DB不可用）
2. 必须恢复DBFunction对象的DB连接，这可能仅在几分钟后才能成功
3. 这意味着必须暂停整个处理，最好是整个管道，以便将传入的消息大量加载到内存中
4. 数据库恢复后恢复处理。处理必须完全恢复与1）遇到问题的消息

现在我知道至少有2个关于故障处理的工具：

1. 卡夫卡消费者抵消
2. apache flink checkpoints

但是，在搜索文档时，我没有看到在单个运算符中如何在流处理过程中使用其中任何一个。

那么，在流应用程序中进行细粒度错误处理和恢复的推荐策略是什么？

Answer 1

几点：

keyBy不会帮助确保按顺序处理。如果有的话，它可以交错来自不同Kafka分区的事件（可能在每个分区内按顺序），从而创建以前不存在的无序。如果不了解您打算使用多少个FlinkKafkaConsumer实例，每个实例将消耗多少个分区，如何在Kafka分区中分配密钥以及您认为的原因，那么很难更具体地评论如何保证有序处理一个keyBy是必要的 - 但如果你正确设置，保持秩序可能是可以实现的。 reinterpretAsKeyedStream在这里很有帮助，但是这个功能很难理解，而且使用起来很棘手。

您可以使用Flink的AsyncFunction以容错，完全一次的方式管理与外部DB的连接。

Flink不支持系统化的细粒度恢复 - 其检查点是整个分布式集群状态的全局快照，旨在作为单一，自洽的快照在恢复期间使用。如果您的作业失败，通常唯一的办法是从检查点重新启动，这将涉及倒带输入队列（存储在检查点中的偏移量），重放自那些偏移的事件，重新发出数据库查找（异步函数）将自动执行），并使用kafka事务实现端到端的一次语义。然而，在令人尴尬的并行工作的情况下，有时可以利用fine-grained recovery。