NiFi实现：为文件每行ID调用REST API（支持并行）

以下是适配NiFi组件特性的具体实现步骤，支持并行处理需求：

核心流程概述 #

读取ID数据源文件 → 将每行ID拆分为独立Flow File → 构建带ID参数的API请求 → 并行调用API → 处理/聚合返回结果

1. 读取ID数据源文件 #

使用GetFile处理器（或ListFile+FetchFile组合）读取存储ID的Flow File，确保文件内每行对应一个独立ID。

2. 拆分单个ID为独立Flow File #

添加SplitText处理器，关键配置：

• Line Split Count：设置为1，让每个Flow File仅包含一个ID，为并行处理创造条件
• 可选配置：若文件带表头，设置Header Line Count跳过表头行

3. 提取ID并构建API请求URL #

1. 用ExtractText处理器将Flow File中的ID内容提取为自定义属性（比如target.id）：
   • 属性规则：target.id = $.*（匹配Flow File全部内容，此时每个文件仅含一行ID）
2. 用UpdateAttribute处理器构建完整API请求URL属性：
   • 添加属性：api.request.url = https://service.com/api/thing/${target.id}

4. 并行调用API #

使用InvokeHTTP处理器完成API调用，关键配置：

• HTTP Method：根据API需求选择GET或对应方法
• Remote URL：填写${api.request.url}，引用之前构建的带参数URL
• 开启并行处理：在处理器Settings标签下，将Concurrent Tasks设为合理数值（如10-20，需匹配API并发限制，避免触发限流）

5. 处理API返回结果 #

• 若API返回JSON数据，可用EvaluateJsonPath提取目标字段，或用ConvertRecord转换为结构化格式
• 如需保留原始响应，直接用PutFile或PutDatabaseRecord将结果写入文件或数据库

6. 错误处理与重试（可选） #

添加RouteOnAttribute处理器，基于InvokeHTTP返回的http.status.code属性路由：

• 规则示例：success = ${http.status.code:matches('2\\d\\d')}，将成功响应路由至后续处理
• failure = ${http.status.code:matches('4\\d\\d|5\\d\\d')}，将失败请求路由至RetryFlowFile处理器重试，或写入失败队列留存排查

7. 聚合结果（可选） #

若需将所有API返回结果合并为单个文件，使用MergeContent处理器：

• 可按文件数量、大小或指定分隔符（如换行符）合并Flow File

内容的提问来源于stack exchange，提问作者Neil P