获取 Nifi PutSQL 处理器的最后执行时间戳

Question

有没有办法通过 REST-API 获取最后一次执行 PutSQL 处理器的时间戳？这样的时间戳是否存在，或者我可以以某种方式自己构建一个？

设置：我有 Airflow 来触发我的 Nifi-ETL，它以几个 PutSQL 处理器结束 - 完成之后，我需要在 Airflow 中执行其他操作。

想法：我想触发第一个 Nifi 处理器，然后在 Airflow 中等待，直到最后一个 PutSQL 处理器的last_execution_timestamp 更新。

问题： 我尝试访问属性statsLastRefreshed，但这不是最后一次执行时间，而是最后一次任何东西（用户/api请求）访问导致Nifi刷新处理器的处理器。

s = processor["status"]["statsLastRefreshed"]  # '13:13:26 CEST'

我在 Airflow 的 REST API 文档中找不到任何内容。

我看到的唯一其他选择是从 Airflow 向最后一个 PutSQL 处理器的数据库表发出请求，以查看那里是否发生了任何新情况。

Answer 1

我想出了一个变通的解决方案。

1. 在处理器中添加一个名为 mypropertyname 的自定义属性，其值为 ${now()}

2. 通过处理器的任何流文件都将具有通过处理器时的时间戳作为属性！

3. 在步骤 1 中的处理器之后有一个 UpdateAttribute 处理器，选项（在处理器属性下）Store State 设置为 Store state locally。

4. 在UpdateAttribute 处理器中添加一个名为readable_property 的自定义属性，并将其设置为值${'mypropertyname'}。

处理器的状态现在包含最后一个流文件的值（例如，带有从步骤 1 开始执行 now() 方法的时间戳）。

4. 通过 REST-API 和 URI /nifi-api/processors/{id}/state 上的 GET（例如在 Airflow 中）获取有状态处理器的值（以及因此通过 (!) 的最后一个流文件的值）

返回的 JSON 包含以下几行：

{
"key":"readable_property"
,"value":"Wed Apr 14 11:13:40 CEST 2021"
,"clusterNodeId":"some-id-0d8eb6052"
,"clusterNodeAddress":"some-host:port-number"
}

然后您只需解析 JSON 以获得 Airflow 中的值。

注意：在前一个处理器将属性添加到带有now 的流文件与流文件实际通过UpdateAttribute 处理器之间会有一点延迟，您可以从中读取时间戳。

Answer 2

Nifi 是为连续数据流而设计的。所以Processor stats 中没有最后执行时间。

虽然 Nifi 能够进行 ETL（未完全成熟），但如果您的流程只需要文件和数据库访问，我相信 Airflow 比 Nifi 更适合 ETL 流程

为 ETL 任务集成 Airflow 和 Nifi 的架构有点复杂。如果可能，您可以考虑选择其中之一，以免遇到您在此处描述的问题。

如果您的流程包含大量不同的输入、复杂的逻辑和摄取大文件，那么仅选择 Airflow 并不容易。