基于流的应用程序中的受控/手动错误/恢复处理 [英] Controlled/manual error/recovery handling in stream-based applications

问题描述

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

I am working on an application based on Apache Flink, which makes use of Apache Kafka for input and out. Possibly this application will be ported to Apache Spark, so I have added this as a tag as well, and the question remains the same.

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

I have the requirement that all incoming messages received via kafka must be processed in-order, as well safely be stored in a persistence layer (database), and no message must get lost.

此应用程序中的流式处理部分相当琐碎/很小，因为主要逻辑将归结为以下内容:

The streaming-part in this application is rather trivial/small, as the main logic will boil down to something like:

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

现在，在此管道中，可能会发生几种错误情况 :

Now, during this pipeline, several error situations could occur:

• 数据库不可用(关闭，表空间已满，...)
• 由于逻辑错误(来自列格式)，无法存储更改
• kafka生产者由于经纪人不可用而无法发送消息

以及其他情况.

现在我的问题是，在这些情况下，我实际上如何必须保证上述内容的一致性，

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

1. Stream-Operator 6) detects a problem (DB unavailable)
2. The DB-connection of the DBFunction object must be recovered, which might only succeed after some minutes
3. This means that overall processing must be suspended, at best for the whole pipeline, so that incoming messages are lot loaded into memory
4. Resume processing after database has been recovered. Processing must resume exactly with the message which encountered the problem at 1)

现在我知道至少有2种有关故障处理的工具:

Now I know that there is at least 2 tools regarding failure handling:

1. kafka消费者抵销额
2. Apache Flink检查点

但是，在搜索文档时，我看不到如何在单个运算符中将它们中的任何一个用于流处理的中间.

However, searching the docs, I fail to see how either of those could be used in the middle of stream processing from within a single operator.

那么，对于流应用程序中的细粒度错误处理和恢复，推荐的策略是什么?

So, what would be the recommended strategies for fine-grained error handling and recovery in a streaming application?

推荐答案

几点:

keyBy不会帮助确保按顺序进行处理.如果有的话，它可能会交错来自不同Kafka分区的事件(可能在每个分区中按顺序排列)，从而在以前不存在的地方造成乱序.在不了解您打算使用多少FlinkKafkaConsumer实例，每个实例将要使用多少个分区，密钥如何在整个Kafka分区中分布以及如何思考的情况下，很难更具体地评论如何保证有序处理keyBy是必要的-但是，如果您正确设置，则可以实现保留顺序. reinterpretAsKeyedStream 会有所帮助此处，但是此功能难以理解，并且很难正确使用.

The keyBy is not going to help ensure in-order processing. If anything, it may interleave events from different Kafka partitions (which may have been in-order within each partition), thereby creating out-of-orderness where it didn't previously exist. It's hard to comment more specifically about how you might guarantee in-order processing without understanding how many FlinkKafkaConsumer instances you intend to use, how many partitions each one will be consuming from, how the keys are distributed across the Kafka partitions, and why you think a keyBy is necessary -- but if you set things up correctly, preserving order may be achievable. reinterpretAsKeyedStream can be helpful here, but this feature is difficult to understand, and tricky to use correctly.

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

You could use Flink's AsyncFunction to manage the connection to the external DB in a fault tolerant, exactly-once, manner.

Flink不以系统的方式支持细粒度的恢复-它的检查点是整个分布式集群状态的全局快照，并且被设计为在恢复期间用作整体的，自洽的快照.如果您的工作失败，通常唯一的办法是从检查点重新启动，这将涉及倒退输入队列(到检查点中存储的偏移量)，从这些偏移量开始重放事件，重新发出数据库查找(异步功能)会自动执行)，并使用kafka事务来一次实现一次端到端的语义.但是，对于尴尬的并行作业，有时可以利用

Flink doesn't support fine-grained recovery in a systematic way -- its checkpoints are global snapshots of the state of the entire distributed cluster, and are designed to be used during recovery as a monolithic, self-consistent, snapshot. If your job fails, normally the only recourse is to restart from a checkpoint, which will involve rewinding input queues (to the offsets stored in the checkpoint), replaying the events since those offsets, re-issuing the DB lookups (which the async function will do automatically), and using kafka transactions to achieve end-to-end exactly once semantics. However, in the case of embarrassingly parallel jobs, it is sometimes possible to take advantage of fine-grained recovery.

这篇关于基于流的应用程序中的受控/手动错误/恢复处理的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！