NiFi - 如何在 ExecuteStreamCommand 中引用流文件? [英] NiFi - how to reference a flowFile in ExecuteStreamCommand?

问题描述

我需要执行以下操作:sed '1d' simple.tsv > noHeader.tsv

这将从我的大流文件(> 1 GB)中删除第一行.

事情是 - 我需要在我的流文件上执行它，所以它是:

sed '1d' myFlowFile > myFlowFile

问题是:我应该如何配置 ExecuteStreamCommand 处理器，以便它在我的流文件上运行命令并将其返回给我的流文件?如果 sed 不是最佳选择，我可以考虑采用其他方式(例如 tail)

谢谢，迈克尔

编辑 2(解决方案):

下面是执行我需要的最终 ExecuteStreamCommand 配置(从流文件中删除第一行).@Andy - 非常感谢所有宝贵的提示.

解决方案

Michal,

我想确保我正确理解了您的问题，因为我认为有更好的解决方案.

问题:

您有一个 1GB 的 TSV 加载到 NiFi 中，并且您想删除第一行.

解决方案:

如果您的文件较小，最好的解决方案是使用

I need to execute something like: sed '1d' simple.tsv > noHeader.tsv

which will remove first line from my big flow file (> 1 GB).

The thing is - I need to execute it on my flow file, so it'd be:

sed '1d' myFlowFile > myFlowFile

Question is: how I should configure the ExecuteStreamCommand processor so that it runs the command on my flow file and returns it back to my flow file? If sed is not a best option, I can consider doing this other way (e.g. tail)

Thanks, Michal

Edit 2 (Solution):

Below is the final ExecuteStreamCommand config that does what I need (remove 1st line from the flow file). @Andy - thanks a lot for all the precious hints.

解决方案

Michal,

I want to make sure I'm understanding your problem correctly, because I think there are better solutions.

Problem:

You have a 1GB TSV loaded into NiFi and you want to remove the first line.

Solution:

If your file was smaller, the best solution would be to use a ReplaceText processor with the following processor properties:

• Search Value: ^.*\n
• Replacement Value: <- empty string

That would strip the first line out without having to send the 1GB content out of NiFi to the command-line and then re-ingest the results. Unfortunately, to use a regular expression, you need to set a Maximum Buffer Size, which means the entire contents need to be read into heap memory to perform this operation.

With a 1GB file, if you know the exact value of the first line, you should try ModifyBytes which allows you to trim a byte count from the beginning and/or end of the flowfile contents. Then you could simply instruct the processor to drop the first n bytes of the content. Because of NiFi's copy-on-write content repository, you will still have ~2GB of data, but it does it in a streaming manner using an 8192B buffer size.

My best suggestion is to use an ExecuteScript processor. This processor allows you to write custom code in a variety of languages (Groovy, Python, Ruby, Lua, JS) and have it execute on the flowfile. Using a Groovy script like the one below, you could remove the first line and copy the remainder in a streaming fashion so the heap does not get unnecessarily taxed.

I tested this with 1MB files and it took about 1.06 seconds for each flowfile (MacBook Pro 2015, 16 GB RAM, OS X 10.11.6). On a better machine you'll obviously get better throughput, and you can scale that up to your larger files.

def flowfile = session.get()
if (!flowfile) return

try {
    // Here we are reading from the current flowfile content and writing to the new content
    flowfile = session.write(flowfile, { inputStream, outputStream ->
        def bufferedReader = new BufferedReader(new InputStreamReader(inputStream))

        // Ignoring the first line
        def ignoredFirstLine = bufferedReader.readLine()

        def bufferedWriter = new BufferedWriter(new OutputStreamWriter(outputStream))
        def line

        int i = 0

        // While the incoming line is not empty, write it to the outputStream
        while ((line = bufferedReader.readLine()) != null) {
            bufferedWriter.write(line)
            bufferedWriter.newLine()
            i++
        }

        // By default, INFO doesn't show in the logs and WARN will appear in the processor bulletins
        log.warn("Wrote ${i} lines to output")

        bufferedReader.close()
        bufferedWriter.close()
    } as StreamCallback)

    session.transfer(flowfile, REL_SUCCESS)
} catch (Exception e) {
    log.error(e)
    session.transfer(flowfile, REL_FAILURE)
}

One side note, in general a good practice for NiFi is to split giant text files into smaller component flowfiles (using something like SplitText) when possible to get the benefits of parallel processing. If the 1GB input was video, this wouldn't be applicable, but as you mentioned TSV, I think it's likely that splitting the initial flowfile into smaller pieces and operating on them in parallel (or even sending out to other nodes in the cluster for load balancing) may help your performance here.

Edit:

I realized I did not answer your original question -- how to get the content of a flowfile into the ExecuteStreamCommand processor command-line invocation. If you wanted to operate on the value of an attribute, you could reference the attribute value with the Expression Language syntax ${attribute_name} in the Arguments field. However, as the content is not referenceable from the EL, and you don't want to destroy the heap by moving the 1GB content into an attribute, the best solution there would be to write the contents out to a file using PutFile, run the sed command against the provided filename and write it to another file, and then use GetFile to read those contents back into a flowfile in NiFi.

Edit 2:

Here is a template which demonstrates using ExecuteStreamCommand with both rev and sed against flowfile content and putting the output into the content of the new flowfile. You can run the flow and monitor logs/nifi-app.log to see the output or use the data provenance query to examine the modification that each processor performs.

这篇关于NiFi - 如何在 ExecuteStreamCommand 中引用流文件?的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！