如何在 Nifi 处理器中提取和操作数据答案

【问题标题】：How to extract and manipulate data within a Nifi processor如何在 Nifi 处理器中提取和操作数据
【发布时间】：2018-06-26 10:30:56
【问题描述】：

我正在尝试编写一个自定义 Nifi 处理器，它将接收传入流文件的内容，对其执行一些数学运算，然后将结果写入传出流文件。有没有办法将传入流文件的内容转储为字符串或其他内容？我一直在寻找一段时间，它似乎并不那么简单。如果有人能向我指出一个很好的教程来处理类似的事情，那将不胜感激。

【问题讨论】：

最好从 ExecuteScript 或 ExecuteGroovyScript 处理器开始。 Groovy 基于 java - 应该很容易。作为处理器的示例，请使用任何简单处理器的源，如下所示：EncodeContent.java

标签： apache-nifi

【解决方案1】：

Apache NiFi Developer Guide 很好地记录了创建自定义处理器的过程。在您的具体情况下，我将从 Component Lifecycle 部分和 Enrich/Modify Content 模式开始。任何其他做类似工作的处理器（如ReplaceText 或Base64EncodeContent）都是值得学习的好例子；所有源代码都可以在GitHub 上找到。

基本上，您需要在处理器类中实现#onTrigger() 方法，读取流文件内容并将其解析为您预期的格式，执行您的操作，然后重新填充生成的流文件内容。您的源代码将如下所示：

    @Override
    public void onTrigger(final ProcessContext context, final ProcessSession session) throws ProcessException {
        FlowFile flowFile = session.get();
        if (flowFile == null) {
            return;
        }

        final ComponentLog logger = getLogger();
        AtomicBoolean error = new AtomicBoolean();
        AtomicReference<String> result = new AtomicReference<>(null);

        // This uses a lambda function in place of a callback for InputStreamCallback#process()
        processSession.read(flowFile, in -> {
            long start = System.nanoTime();

            // Read the flowfile content into a String
            // TODO: May need to buffer this if the content is large
            try {
                final String contents = IOUtils.toString(in, StandardCharsets.UTF_8);
                result.set(new MyMathOperationService().performSomeOperation(contents));

                long stop = System.nanoTime();
                if (getLogger().isDebugEnabled()) {
                    final long durationNanos = stop - start;
                    DecimalFormat df = new DecimalFormat("#.###");
                    getLogger().debug("Performed operation in " + durationNanos + " nanoseconds (" + df.format(durationNanos / 1_000_000_000.0) + " seconds).");
                }
            } catch (Exception e) {
                error.set(true);
                getLogger().error(e.getMessage() + " Routing to failure.", e);
            }
        });

        if (error.get()) {
            processSession.transfer(flowFile, REL_FAILURE);
        } else {
            // Again, a lambda takes the place of the OutputStreamCallback#process()
            FlowFile updatedFlowFile = session.write(flowFile, (in, out) -> {
                final String resultString = result.get();
                final byte[] resultBytes = resultString.getBytes(StandardCharsets.UTF_8);

                // TODO: This can use a while loop for performance
                out.write(resultBytes, 0, resultBytes.length);
                out.flush();
            });
            processSession.transfer(updatedFlowFile, REL_SUCCESS);
        }
    }

Daggett 说得对，ExecuteScript 处理器是一个很好的起点，因为它会缩短开发生命周期（无需构建 NAR、部署和重新启动 NiFi 来使用它），并且当你有正确的行为时，你可以轻松地复制/粘贴到生成的骨架中并部署一次。

【讨论】：

我很好奇为什么 Nifi 开发者指南使用 OutputStreamCallback.process() 函数，如果它可以被 lambda 替换？是否有无法使用 lambda 的原因或情况？
开发者指南最初是在 Java 8 被支持（甚至存在）并且 lambdas 可用之前编写的。更冗长的行为仍然有效，并且可以说明对于不熟悉 lambda 的开发人员正在发生的事情。文档被视为代码，始终欢迎社区改进。