如何在 nifi 处理器中获取整个流文件内容答案

【问题标题】：How to take Entire flowfile content in nifi processor如何在 nifi 处理器中获取整个流文件内容
【发布时间】：2018-08-26 03:34:28
【问题描述】：

我正在使用 nifi 来开发数据漂移。在我使用 SelectHiveQL 处理器的流程中。 selectHiveQL 的输出（flowFile）需要进入下一个处理器。什么是合适的处理器来获取 flowFile 内容并存储到用户定义的变量中必须在 Executescript 中使用相同的变量来操作数据。

【问题讨论】：

flowFile.read() 返回输入流
有很多方法可以将整个内容加载到属性中，但它会占用整个内存，并且可能导致内存不足错误/流文件处理速度变慢的可能性。这样您就可以从该流文件中获取所需的部分并将其处理为更好的方式。\
感谢我们在流文件中描述的表输出不超过 1kb 数据，我想在 uodateattribute 处理器中获取流文件内容。

标签： apache-nifi

【解决方案1】：

ExecuteScript 处理器可以通过标准 API 直接访问传入流文件的内容。这是一个例子：

def flowFile = session.get();
if (flowFile == null) {
    return;
}

// This uses a closure acting as a StreamCallback to do the writing of the new content to the flowfile
flowFile = session.write(flowFile,
        { inputStream, outputStream ->
            String line

            // This code creates a buffered reader over the existing flowfile input
            final BufferedReader inReader = new BufferedReader(new InputStreamReader(inputStream, 'UTF-8'))

            // For each line, write the reversed line to the output
            while (line = inReader.readLine()) {
                outputStream.write("${line.reverse()}\n".getBytes('UTF-8'))
            }
        } as StreamCallback)

flowFile = session?.putAttribute(flowFile, "reversed_lines", "true")
session.transfer(flowFile, /*ExecuteScript.*/ REL_SUCCESS)

将流文件内容移动到属性是危险的，因为属性和内容内存在 NiFi 中的管理方式不同。 Apache NiFi In Depth 指南中对差异有更详细的解释。

【讨论】：

【解决方案2】：

您可以使用 ExtractText 将流文件的内容提取到属性中。

在 ExtractText 处理器中，您将创建一个属性（您为该属性指定的名称将是流文件中的一个新属性），该属性的值将是正则表达式 (\A.+\Z)。根据我的经验，这个正则表达式足以捕获流文件的全部内容，但我认为里程可能会因流文件中的内容类型而异。

【讨论】：

感谢乔的回答。虽然你是正确的，这是可能的，但我会提醒你确保内容大小非常小（