Nifi: Is it possible to use preveious flowfile data for transfering new one?

[sally <sa...@gmail.com>]

I have several flowfiles in sucess queue in which i have parameter count and
i have to transfer only counter amount flowfiles , every time counter value
is different so any time i will have to get it from flowfile( all flowfile
from this success queue have this attribute), i have tried this logic below
( i mean my code) and several modifications of it but i can't fulfill this
task, could you tell me what's wrong with this code?

import org.apache.commons.io.IOUtils
import java.nio.charset.StandardCharsets
import groovy.lang.*
 
def flowFile= session.get();
if (!flowFile) return;
int numb=(flowFile.getAttribute("count") as Double).round()
def filename=flowFile.getAttribute("filename")
 
 
def flowfileList = session.get(numb-1)
 
if(flowfileList.size() == numb-1) {
session.transfer(flowfile,REL_SUCCESS);
session.transfer(flowfileList,REL_SUCCESS);
 
}
else{
session.rollback()
 
}



--
Sent from: http://apache-nifi-users-list.2361937.n4.nabble.com/

Re: Nifi: Is it possible to use preveious flowfile data for transfering new one?

[Juergen Boiselle <jb...@icloud.com>]

Hi Sally,

I guess your three questions relate to each other, right?

So, you get from one incoming queue FlowFiles with an Attribute “count” which tells you, how much more FlowFiles in the same queue you have to read and process together, right?

NIFI is highly parallelised and all components work as independently as possible. Therefore your FlowFiles are evenly spread over all threads of one processor and even for one thread you can not say, which FlowFile will be next in your queue as they’re are taken by “fastest one first”. Especially you cannot say, that the >first< FlowFile contains some information about the >next< FlowFiles as FlowFiles are not in a special order in the queue.

But what can you try?

I’d suggest to try with one or more standard processors available. My best guess is MergeContent to put FlowFiles together and then take a second Processor (maybe some scripted one to do the processing). When doing your tests with MergeContent, make sure, that you have only one thread running, as this helps to understand the behaviour better: In “configure” of MergeContent, on the “Scheduling” Tab, set “Concurrent Tasks” to 1. If you’re in a NIFI-cluster environment also set it to run on “Primary Node”.

The MergeContent processor can handle incoming FlowFiles and bucket them together based on attributes in the FlowFiles.

One little thing about MergeContent that confuses a lot of people but has a simple work around: Make sure, it has only one incoming queue. This is also true, if you want to redirect failure back into MergeContent. In such a case put a Funnel in front of MergeContent and connect all incoming queues (including failure) into the Funnel. It then works as expected and documented.

cheers, Jboi

> On 28. Oct 2017, at 21:52, sally <sa...@gmail.com> wrote:
> 
> I have several flowfiles in sucess queue in which i have parameter count and
> i have to transfer only counter amount flowfiles , every time counter value
> is different so any time i will have to get it from flowfile( all flowfile
> from this success queue have this attribute), i have tried this logic below
> ( i mean my code) and several modifications of it but i can't fulfill this
> task, could you tell me what's wrong with this code?
> 
> import org.apache.commons.io.IOUtils
> import java.nio.charset.StandardCharsets
> import groovy.lang.*
> 
> def flowFile= session.get();
> if (!flowFile) return;
> int numb=(flowFile.getAttribute("count") as Double).round()
> def filename=flowFile.getAttribute("filename")
> 
> 
> def flowfileList = session.get(numb-1)
> 
> if(flowfileList.size() == numb-1) {
> session.transfer(flowfile,REL_SUCCESS);
> session.transfer(flowfileList,REL_SUCCESS);
> 
> }
> else{
> session.rollback()
> 
> }
> 
> 
> 
> --
> Sent from: http://apache-nifi-users-list.2361937.n4.nabble.com/