我有类似的 nifi 流（独立）
executestreamprocessor(hive script) -> executestreamprocessor(hadoop script).
对于每个传入的流文件，hive 脚本使用命令 INSERT..INTO..SELECT..FROM 运行，hadoop 脚本从存储区域中删除特定文件。
有时，当 hadoop 脚本同时运行命令时，hive 脚本会失败。
我每小时最多可以获得 4 个文件。所以我计划在 hive 和 hadoop 处理器之间使用 controlrate 处理器。我设置了队列数量达到4个流文件时的条件，然后应该执行hadoop脚本。但是，controlrate 具有仅设置最大速率的属性。它没有最低费率。
是否有任何可能的解决方案来实现？或任何其他解决方案？
【问题讨论】：
标签： apache-nifi
您应该可以为此使用ExecuteScript，试试这个 Groovy 脚本：
def flowFiles = session.get(4)
if(!flowFiles || flowFiles.size() < 4) {
session.rollback()
} else {
session.transfer(flowFiles, REL_SUCCESS)
}
如果您只想触发一次下游流，那么您可以从父级创建一个子流文件（并报告一个 JOIN 出处事件）：
def flowFiles = session.get(4)
if(!flowFiles || flowFiles.size() < 4) {
session.rollback()
} else {
def flowFile = session.create(flowFiles)
session.provenanceReporter.join(flowFiles, flowFile)
session.remove(flowFiles)
session.transfer(flowFile, REL_SUCCESS)
}
话虽如此，如果您不关心流文件内容（即，您使用流文件作为触发器），您可以使用 MergeContent，最小和最大条目数 = 4。
【讨论】：