使用 sys.stdout.write() 在 NiFi 中创建多个文件？

Question

我在 NiFi 中有一个管道，可以提取一些我需要清理的无效 JSON。我设计的最佳解决方案是通过ExecuteStreamCommand 运行 Python 脚本，同时一举清理/拆分它。然而，即使我在 for 循环中使用 sys.stdout.write()，在 NiFi 的输出流中也只有原始 JSON 出现。

我是在滥用sys.stdout.write() 还是这可能，但我只是做错了什么？我的最终目标是让json的每一行都成为一个新的流文件，即文件1是{"fruit":"apple",...，文件2是{"fruit":"cherry",...，依此类推。

示例 JSON

{"fruit":"apple", "vegetable":"celery", "location":{"country":"nor\\way", "city":"oslo", }, "color":"blue"}
{"fruit":"cherry", "vegetable":"kale", "location":{"country":"france", "city":"calais", }, "color":"green"}
{"fruit":"peach", "vegetable":"peas", "location":{"country":"united\\kingdom", "city":"london", }, "color":"yellow"}

脚本

import json
import re
import sys

flow_file = sys.stdin.read()
try:
    load = json.loads(flow_file)
    sys.stdout.write(flow_file)
except:
    flow_file_esc = re.sub(r"[(\\)]", "", flow_file)
    for f in flow_file_esc.splitlines():
        sys.stdout.write(str(f))

Answer 1

你可以先用 ReplaceText 清理文件，然后用 SplitJson、SplitRecord 或 ForkRecord 拆分它吗？

如果您需要结合这两个操作并想要编写脚本，您可以尝试使用 Jython 执行 ExecuteScript（因为它看起来不像您使用的是本地 CPython 库），我在 my cookbook 中有一些简单的示例和my blog.