如何从QueryMarkLogic返回的NiFi FlowFile运行XQuery并标记ML已处理文档

核心思路 #

通过NiFi的MarkLogic处理器链，先获取待迁移文档，提取文档URI后执行XQuery为文档添加Processed集合，后续查询过滤已标记文档，避免重复处理。

具体步骤 #

1. 确认QueryMarkLogic输出的FlowFile包含URI属性 #

QueryMarkLogic处理器查询MarkLogic文档后，会自动将文档URI写入FlowFile的ml.uri属性（默认行为），无需额外配置，确保查询逻辑能正确返回目标文档即可。

2. 配置ExecuteScriptMarkLogic处理器 #

该处理器用于执行添加集合的XQuery操作，关键配置如下：

• Script Type：选择XQuery
• Script：填入以下代码，通过NiFi内置的$nifi对象获取FlowFile的ml.uri属性，调用MarkLogic函数标记文档：

let $target-uri := $nifi:getAttribute('ml.uri')
return xdmp:document-add-collections($target-uri, "Processed")

• MarkLogic Connection Pool Service：选择与QueryMarkLogic相同的ML连接池（需确保连接池的用户账号拥有文档更新权限）
• 其余配置保持默认即可

3. 编排处理器流程 #

• 将QueryMarkLogic的success输出关系连接到ExecuteScriptMarkLogic的input端口
• ExecuteScriptMarkLogic的success输出连接到PutAzureBlobStorage（或其他Azure Blob写入处理器）完成迁移；failure输出可连接到重试队列或告警处理器

4. 优化QueryMarkLogic查询避免重复处理 #

修改QueryMarkLogic的查询语句，过滤已加入Processed集合的文档，示例XQuery：

(: 仅返回未标记为Processed的文档 :)
fn:collection()[not(xdmp:document-get-collections(fn:document-uri(.)) contains "Processed")]

若使用Query Builder模式，可在过滤条件中添加「集合不包含Processed」规则。

内容的提问来源于stack exchange，提问作者XCELERENT - I want to dance