NiFi 上的 Python 2.X：json.loads 中的Ñ（和其他）问题

Question

我正在使用 Jython InvokeScriptedProcessor 将数据从 json 结构构造到 sql 结构。我在使用特定功能时遇到问题。 json.loads。 json.loads 无法识别特殊字符，如 ñ、é、á、í...

它以一种奇怪的形式写出来。而且我还没有达到任何形式来拥有它。

例如（很简单）

{"id":"ÑUECO","value":3.141592,"datetime":"....","location":"ÑUECO"}

如果我们试着用 sql 来写

INSERT INTO .... (id, value) VALUES ("...",3.141592);

它会失败。它让我失望。我无法通过任何返回选项返回数据，成功或失败，NiFi 的版本无关紧要。这是我的代码

    def process(self, inputStream, outputStream):
        # read input json data from flowfile content
        text = IOUtils.toString(inputStream, StandardCharsets.UTF_8)
        data = json.loads(text) 

都没有

        data = json.loads(text.encode("utf-8"))

正常工作。文本采用 unicode 格式。

                def __generate_sql_transaction(input_data):
                    """ Generate SQL statement """

                    sql = """
        BEGIN;"""

                    _id = input_data.get("id")
                    _timestamp = input_data.get("timestamp")
                    _flowfile_metrics = input_data.get("metrics")
                    _flowfile_metadata = input_data.get("metadata")

                    self.valid = __validate_metrics_type(_flowfile_metrics)

                    if self.valid is True:
                        self.log.error("generate insert")
                        sql += """
            INSERT INTO
                {0}.{1} (id, timestamp, metrics""".format(schema, table)

                        if _flowfile_metadata:
                            sql += ", metadata"
                        sql += """)
            VALUES
                ('{0}', '{1}', '{2}'""".format(_id.encode("utf-8"), _timestamp, json.dumps(_flowfile_metrics))

                        self.log.error("generate metadata")
                        if _flowfile_metadata:
                            sql += ", '{}'".format(json.dumps(_flowfile_metadata).encode("utf-8"))
                        sql += """)
            ON CONFLICT ({})""".format(on_conflict)

                        if not bool(int(self.update)):
                            sql += """
                DO NOTHING;"""
                        else:
                            sql += """
                DO UPDATE
                    SET"""
                            if bool(int(self.preference)):
                                sql += """
                        metrics = '{2}' || {0}.{1}.metrics;""".format(schema, table, json.dumps(_flowfile_metrics))
                            else:
                                sql += """
                        metrics = {0}.{1}.metrics || '{2}';""".format(schema, table, json.dumps(_flowfile_metrics))

                    else:
                        return ""

                    sql += """
        COMMIT;"""
                    return sql

我再次将数据发送到 NiFi：

        output = __generate_sql_transaction(data)
        self.log.error("post generate_sql_transaction")
        self.log.error(output.encode("utf-8"))

        # If no sql_transaction is generated because requisites weren't met,
        # set the processor output with the original flowfile input.
        if output == "":
            output = text

        # write new content to flowfile
        outputStream.write(
            output.encode("utf-8")
        )

那个输出看起来像

INSERT INTO .... VALUES ("ÃUECO","2020-01-01T10:00:00",'{"value":3.1415}','{"location":"\u00d1UECO"}');

我在元数据中也有“Ñueco”，但它不适用于 id 或元数据

注意：似乎 InvokeScriptedProcessor 使用 Groove 而不是 Python 可以正常工作。但我的问题是我对 Groovy 一无所知...

有没有人发现类似的问题？你是怎么解决的？

更新：

输入示例：

{"id":"ÑUECO",
 "metrics":{
     "value":3.1415
 },
 "metadata":{
     "location":"ÑUECO"
 },
 "timestamp":"2020-01-01 00:00:00+01:00"
}

期望的输出：

BEGIN;
INSERT INTO Table (id, timestamp, metrics, metadata)
VALUES ('ÑUECO', 
        '2020-01-01T00:00:00+01:00',
        '{"value":3.1415}',
        '{"location":"ÑUECO"}')
ON CONFLICT (id, timestamp)
DO UPDATE
    SET
        metrics='{"value":3.1415}' || Table.metrics;
COMMIT;

实际输出：

BEGIN;
INSERT INTO Table (id, timestamp, metrics, metadata)
VALUES ('ÃUECO', 
        '2020-01-01T00:00:00+01:00',
        '{"value":3.1415}',
        '{"location":"\u00d1UECO"}')
ON CONFLICT (id, timestamp)
DO UPDATE
    SET
        metrics='{"value":3.1415}' || Table.metrics;
COMMIT;

Answer 1

UPD

1. jython 不能正确处理字节字符串 - 所以，不要使用 .encode('utf-8')

2. 使用java方法将内容写回具有特定编码的流文件

下面是一个正确读写非ascii字符的示例，包括Ñ

将ExecuteScript 处理器与jython 一起使用并替换_transform(text) 函数的主体：

import traceback
from org.apache.commons.io import IOUtils
from java.nio.charset import StandardCharsets
from org.apache.nifi.processor.io import StreamCallback

class FlowWriter(StreamCallback):
    def _transform(self, text):
        # transform incoming text here
        return '@@@@' + text + '****'

    def process(self, inputStream, outputStream):
        text = IOUtils.toString(inputStream, StandardCharsets.UTF_8)
        new_text = self._transform(text)
        IOUtils.write(new_text, outputStream, StandardCharsets.UTF_8)

flowFile = session.get()
if flowFile != None:
    try:
        flowFile = session.write(flowFile, FlowWriter())
        flowFile = session.putAttribute(flowFile, "filename", 'headerfile.xml')
        session.transfer(flowFile, REL_SUCCESS)
        session.commit()
    except Exception as e:
        log.error("{}\n{}".format(e,traceback.format_exc()))
        session.rollback(True)  # put file back and penalize it

Answer 2

我最近找到了这个答案。

https://stackoverflow.com/a/35882335/7634711

这不是 NiFi 的问题。这是Python2 的问题以及它如何与json 库一起使用。如果字典键中有特殊字符，问题也会出现在Python3。