How to Merge Multiple Flow Files - "ExecuteSQL"

[#1]

Good morning everybody,

I have the following scenario of returning an API through two streams =

Flow 1 (

{"cod": 1}

{"cod": 2}
{"cod": 3})

Flow 2 (
{"error": "error 1"}
{"error": "error 2"}
{"error": "error 3"})

Each error has its cod and I need to send it to my database in this way through a procedure through "ExecuteSQL" =

procedure (cod => '$ {cod}',
                  error => '$ {error}');

I am trying to do this through "MergeContent" but it is not working.

What Nifi is doing is as follows =

{"cod": 1} {"cod": 2} {"cod": 3} {"error": "error 1"} {"error": "error 2"} {"error": "error 3" }

Not dividing the shipments by code, let's say and passing to "ExecuteSQL" all at once, when Cod + Error would have to be sent one at a time.

Follow all my flow =

This is my "MergeContent" = 

This is my "ExecuteSQL" =

Can someone help me?

[#2]

Good morning all,

Can someone help me?

[#3]

@MattWho  @pvillard @mattw  

Please can you help me?

[#4]

@mburgess can You help me?

[#5]

@leandrolinof 

The NiFi merge processor is working exactly as designed.  Binary concatenation simply appends the content from one FlowFile to the end of the previous FlowFIle's content.

You are trying to perform a special formatted merge.  There is no way to configure the mergeContent processor with custom merge logic.

I am not clear on the what the content looks like at the various stages of your dataflow.  You mention merging FlowFiles from two streams; however, the flow screenshot you shared as 3 flow streams leading to the MergeContent processor.

Are you saying the content of a FlowFile down one path is exactly this:

Flow 1 (
{"cod": 1}
{"cod": 2}
{"cod": 3})

and on the other path it is exactly this:

Flow 2 (
{"error": "error 1"}
{"error": "error 2"}
{"error": "error 3"})

or is it just the json objects without the "flow <num> ( )" wrapped around it?
I am guessing above is what is extracted from the invokeHTTP response content?

What does the FlowFile content look like on that third path before it reaches MergeContent?
I might be helpful with examples of content post InvokeHTTP on all three paths and then what the exact mergeContent based on that example you would want to see.

What logic can be get from the above two flows that tells us the first "cod" goes with the first "error"? 
So if we were to split each into three separate FlowFiles, would there be a way to determine what cod should go with what error?


The ExecuteSQL processor configuration you shared is using NiFi Expression Language (NEL).  NEL does not read from a FlowFiles content.  ${cod} and ${erros} will looks for attributes with these names created on FlowFile and replace that with the values assigned to those attributes.  If those FlowFile Attributes do not exist on the FlowFile, NEL will check the variable registry for those attribute strings.

Hope this helps give you some direction with your dataflow design journey here.
Matt

[#6]

@MattWho,

Good afternoon.

I changed my flow to make a counter by the processor "RouteOnAttribute 1.12.1" so every time I have a record to be sent to the API and have a return to store in Oracle, I do the LOOP through RouteOnAttribut. Thus sending one FlowFile at a time to the PROCEDURE call in Oracle by "ExecuteSQL 1.12.1"

But what is happening at the moment is that from the second time it goes through the LOOP, it seems to me that the Flow to get the attributes "CODIGO" and "JSON_ORIGINAL" is not passing the values ​​that I requested in the processor "EvaluateJsonPath 1.12.1" passing the parameters as null.

Do you have any idea what may be happening in this second flow passage within the LOOP?

This is the JSON ORIGINAL of the second stream that I must send to Oracle = 

This is OK with the CODE attribute as you can check.

This is Split Configure Processor =

This is EvaluateJsonPath 1.12.1 Configure Processor =

This is MergeContent 1.12.1 Configure Processor =

This is ExecuteSQL 1.12.1 Configure Processor =