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

Question

我试图编写一个自定义的Nifi处理器，它将接收传入流文件的内容，对其执行一些数学运算，然后将结果写入传出流文件。是否有一种方法可以将传入流文件的内容转储到字符串或其他东西中？我已经寻找了一段时间，现在似乎没有那么简单。如果有人能为我指出一个很好的教程，处理这样的事情，它将非常感谢。

Answer 1

Apache NiFi开发人员指南非常好地记录了创建自定义处理器的过程。在您的具体案例中，我将从组件生命周期部分和Enricg/Modify Content模式开始。任何其他执行类似工作的处理器（如ReplaceText或Base64EncodeContent）都将是学习的好例子；所有的源代码都可以在GitHub上获得。

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

    @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来使用它），并且当您有正确的行为时，您可以轻松地复制/粘贴到生成的框架中并部署一次。