NiFi中ExecuteStreamCommand输出JSON处理及格式转换问题求助

问题核心原因 #

1. 内容类型误识别：ExecuteStreamCommand默认将脚本输出识别为application/avro-binary，但你的Python脚本实际输出的是JSON字符串，导致后续处理器无法正确解析内容。
2. AttributesToJson使用场景错误：该处理器是把FlowFile的属性转为JSON，而非将FlowFile的内容提取为属性或JSON，因此你看到只有属性名称无对应值。

正确处理流程 #

步骤1：修正ExecuteStreamCommand的输出内容类型 #

在ExecuteStreamCommand的配置中，手动设置Output Content Type为application/json，强制指定输出为JSON格式，避免NiFi自动误识别。

步骤2：提取JSON键值对（按需选择） #

方案A：将JSON内容转为FlowFile属性

使用JsonPathReader处理器：

• 配置Reader为JsonPathReader
• 在JsonPath Expressions中添加需要提取的键值映射，例如：
  • $.username 对应属性名user_name
  • $.email 对应属性名user_email
• 完成配置后，JSON内容里的字段会被提取为FlowFile属性，供后续处理器调用。

方案B：直接用JSON内容对接PutDatabaseRecord

无需转属性，直接写入数据库：

1. 确保ExecuteStreamCommand输出内容类型为application/json
2. 配置PutDatabaseRecord处理器：
   • Record Reader选择JsonTreeReader
   • Record Writer选择对应数据库的写入器（如DatabaseRecordSetWriter）
   • 配置数据库连接池及字段与表列的映射关系，即可直接将JSON内容写入数据库。

额外排查点 #

• 检查Python脚本输出：确保json.dump是输出到标准输出（stdout），而非本地文件，因为ExecuteStreamCommand仅读取脚本的标准输出作为FlowFile内容。
• 验证FlowFile内容：用ViewContent处理器查看ExecuteStreamCommand的输出，确认是有效的JSON字符串。

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