如何根据 SQL 查询的结果过滤流文件？

Question

是否可以根据返回单行结果的 SQL 查询的结果来路由流文件？例如，如果结果为“1”，则将处理流文件；否则会被忽略。

解决方案

以下方法最适合我。

1. 使用 ExecuteSQL 处理器来运行过滤 SQL 查询。按照 Shu 建议的方式编写查询以生成单个记录（匹配）或空记录集（不匹配）。
2. 将 ExecuteSQL 连接到 RouteOnAttribute 处理器，以便使用以下路由属性值 ${executesql.row.count:replaceNull(0):gt(0)} 过滤掉不匹配的流文件

注意，应用ExecuteSQL后，流文件的原始内容会丢失。在我的情况下这不是问题，因为我在处理流文件内容之前进行过滤，并且我的 SQL 查询完全基于流文件属性而不是其内容。虽然在更一般的情况下，当流的传入部分修改流文件内容时，应该将文件内容保存在某处（例如文件系统）并在过滤部分应用后将其恢复。

Answer 1

您可以在您的 sql 查询 where <field_name> = 1 中添加 where 子句，那么我们只会在 result value =1 时输出 flowfile强>。

（或）

检查 NiFi 中的数据：

我们将获得 AVRO 格式的数据作为 SQL 查询的结果，以便您可以使用

option1:ConvertAvroToJson 处理器：

将 AVRO 数据转换为 JSON 格式，然后使用 EvaluateJsonPath 处理器从 json 内容中提取值作为属性。

然后使用 RouteOnAttribute 处理器添加新属性，使用 NiFi expression language equals function 比较值并将流文件路由到匹配关系。

有关 EvaluateJsonpath 和 RouteOnAttribute 处理器配置的更多详细信息，请参阅 this 链接。

选项 2：使用 QueryRecord 处理器：

通过使用 QueryRecord 处理器，我们可以对流文件的内容运行 SQL 查询

将新属性添加到处理器

select * from FLOWFILE where <filed_name> =1 

将属性关系提供给其他处理器

有关 QueryRecord 处理器使用的更多详细信息，请参阅 this 链接。