apache-nifi - 调试复杂 NiFi 数据流的理想方式

Question

根据我在使用 NiFi 构建一些数据库摄取 PoC 后的理解，整个数据流作为流文件流运行。并且在任何特定时间，执行控制可以同时在一个或多个处理器上。

所以我真的很困惑如何针对任何故障调试复杂的数据流。

我的 PoC 工作流程本身如下所示。

当我们处理生产用例时，它会变得比这复杂得多。所以我没有什么问题。

1. 如何知道数据流的状态。如果假设 10 个 fork 流文件中有 4 个在 GenerateTableFetch 中因数据库池错误而失败，我如何知道哪些文件失败以及如何快速重放它们而无需逐个查看数据来源并逐一执行。

2. 有没有一种方法可以仅通过查看数据流来了解哪个流文件在哪个处理器上出现故障。

我对使用 NiFi 调试数据流有很多疑问/困惑，如果有人可以指点我一些文档或分享最佳实践，那会很有帮助。

谢谢。

Answer 1

1- How to know the status of the dataflow. If let's say 4 out 10 forked flow files failed at GenerateTableFetch for database pool error, how do I know which ones failed and how to quickly replay them without going to data governance and doing one by one.

这可以通过将失败类型或任何其他类型的关系发送到进程组来处理错误来管理，具体取决于您使用的处理器类型。

所以就像 Bryan 提到的那样，除非您不在乎，否则您不希望它们自动终止。

2- Is there a way to know just by looking at the dataflow that which flowfiles at which processor are failing.

是的——你必须设置“公告级别”来区分日志级别

如何管理失败的 NiFi 流程？

好吧，您需要与公告板成为最好的 friend ，请参阅此处 SiteToSiteStatusReportingTask 或者您可以使用 InvokeHttp 对 native NiFI Rest Api 进行 GET 调用 http://nifi-server:port/nifi-api/flow/bulletin-board 并且这将响应一个详细的 json 对象，该对象可以被解析，然后推送到 PutSlack/PutEmail/PutSNS 以解决任何错误。

拥有Shared Process Group 来处理任何传入的错误流文件也是理想的选择，此 PG 将使用规则和路由构建，以应用于 NiFi 服务器中的所有数据流逻辑。拥有 PG 特定属性至关重要，这些属性将随您的所有流一起携带，并将在数据流的整个过程中使用。

例如:

进程组“Demo”有一个名为Set PG Attributes的处理器，它设置了PGName属性，PGType属性，FailEmailTitle 属性等。如果我的流程在任何时候失败，失败关系将根据 Set PG Attributes 处理器

中设置的属性之一的值来路由我失败的流程

这是我当前设置的图表，其中我将所有故障发送到同一个共享 PG。

其他选项

如果您认为公告仅持续 5 分钟是个问题，那么您可以使用 nifi-app.log，它可以设置为由 /中的规则填充选择/nifi/conf/logback.xml 文件

  <logger name="org.apache.nifi" level="ERROR"/>
    <logger name="org.apache.nifi.processors" level="DEBUG"/>
    <logger name="org.apache.nifi.processors.standard.LogAttribute" level="ERROR"/>
    <logger name="org.apache.nifi.processors.standard.LogMessage" level="ERROR"/>
    <logger name="org.apache.nifi.controller.repository.StandardProcessSession" level="ERROR" />

因此，您可以让 tailFile 处理器查看您的本地日志文件并获取错误信息或您认为对您有用的信息。