我有一个 JSON 流文件，我需要确定我应该执行 INSERT 还是 UPDATE。诀窍是只更新与 JSON 属性匹配的列。我有一个 ExecuteSQL 工作，它返回 executesql.row.count，但是我丢失了我计划用作 routeonattribute 的原始 JSON 流文件。我试图让 MergeContent 加入 ExecuteSQL（转储 Avro 输出，我只需要 executesql.row.count 属性）和 JSON 流。在执行 ExecuteSQL 之前我已经设置了跟随：
fragment.count=2
fragment.identifier=${UUID()}
fragment.index=${nextInt()}
如果有办法循环遍历与 Oracle 表匹配的 JSON 属性列表，我也可以创建一个 MERGE？
【问题讨论】：
标签： apache-nifi
您的 JSON 有多大？如果它很小，您可以考虑使用 ExtractText（匹配整个文档）将 JSON 转换为属性。然后您可以运行 ExecuteSQL，然后 ReplaceText 将 JSON 放回内容中（覆盖 Avro 结果）。如果您的 JSON 很大，您可以设置 DistributedMapCacheServer 并（在单独的流程中）运行 ExecuteSQL 并将值或 executesql.row.count 存储到缓存中。然后在 JSON 流中，您可以使用 FetchDistributedMapCache 并设置“将缓存值放入属性”属性集。
如果你只需要JSON来使用RouteOnAttribute，也许你可以在ExecuteSQL之前使用EvaluateJsonPath，这样你的条件已经在属性中了，你可以替换流文件内容。
如果你想使用 MergeContent，你可以将 fragment.count 设置为 2，而不是使用 UUID() 函数，你可以使用 UpdateAttribute 将 "parent.identifier" 设置为 "${uuid}"，然后 @987654324 @ 创建 2 个副本，然后 UpdateAttribute 将“fragment.identifier”设置为“${parent.identifier}”，将“fragment.index”设置为“${nextInt():mod(2)}”。这提供了一组可合并的两个流文件，您可以在 fragment.index 为 0 或 1 上进行路由，将一个发送到 ExecuteSQL，一个通过另一个流，在 MergeContent 处加入。
另一种选择是使用设置为“UPDATE”的ConvertJSONToSQL，如果失败，将这些流文件路由到另一个设置为“INSERT”的 ConvertJSONToSQL 处理器。
【讨论】：