如何等待 GenerateTableFetch 查询完成

Question

我的用例是这样的。我有一些 X 表要从 MySQL 中提取。我使用SplitText 拆分它们，将每个表放在一个单独的流文件中，并使用GenerateTableFetch 和ExecuteSQL 拉取。

我希望在所有表的导入完成后收到通知或采取其他措施。在SplitText 文本处理器，我已将original 关系路由到${filename} 上的Wait，目标计数${fragment.count}。这将跟踪完成了多少张表。

但现在我无法弄清楚如何知道特定表何时完成。 GenerateTableFetch 根据分区大小将流文件分成多个。但它没有写入像 fragment.count 这样的属性，我可以用它来等待每个表。

有什么方法可以实现吗？或者有没有办法在整个流程结束时知道流程中的所有流程文件是否都已处理，并且队列中没有任何内容或正在处理？

Answer 1

如果你有一个独立的 NiFi 实例（或者没有在集群之间将流文件分发到 ExecuteSQL 节点），那么你可以使用 QueryDatabaseTable 代替，它（默认情况下）只会在整个结果集时发布所有流文件被处理。如果您将所有行放入单个流文件中，则流文件已被传输到下游这一事实表明获取已完成。

我写了NIFI-5601 来介绍对GTF生成的流文件添加fragment.*属性的改进。

Answer 2

在 NiFi 添加对此的支持之前，我设法使用MergeContent 使其工作。使用 table_name 作为Correlation attribute name，然后使用merged 与Wait 处理器的关系，使用${merge.count} 作为目标。如果有人想要这样做，请参考屏幕截图。