NiFi Processor to Dynamically Create SQL Query From FlowFile or JSon

[#1]

Hi all,

First of all I am a NiFi newbie! 🙂

We are STUCK.

Our initial desire is to get a attribute from a flowfile and asks the SQLServer if this attribute exists. If the answer is YES, then the flowfile continue his way - but we didn't find anything like this.

Then we had an idea: Query the database and put the Attributes in the SELECT. But, again, we didn't find a processor that create a SQL Query dynamically.

The flowfile attribute looks like this:

[ {
  "ID" : 14733307,
  "DeviceID" : "40d71f9776cbdca1a39b6f8c5493219f16ce5e8ac77a537f3936d5b81c367320",
  "LogDate" : "2017-07-03 12:27:26.327",
  "Latitude" : -29.98709106445312,
  "Longitude" : -51.1904296875
} ]

And we are trying to get this attributes and create the query like the one below:

SELECT 
  '${ID}' as ID
  ,'${DeviceID}' as DeviceID
  ,'${LogDate}' as LogDate
  ,'${Latitude}' as Latitude
  ,'${Longitude}' as Longitude
  ,'OK' as OK
FROM [Device] 
WHERE DeviceID = '${DeviceID}'

Is there any Processor that do this? Or we will have to create one? 😄

Cause the examples that we find searching in internet only demonstrate how to replace text and using a lot of processors. Like the two below:

https://stackoverflow.com/questions/41057483/how-to-pass-values-dynamically-in-apache-nifi-from-exec...

https://community.hortonworks.com/questions/70562/executesql-dynamic-query.html

This is working:

What we want:

Any help will be appreciated! 😄

[#2 · Accepted solution]

Hello @Gabriel Queiroz

I hope I understood your use-case properly. How about using EvaluateJsonPath to extract each values into attributes so that ExecuteSQL can refer?

A template file for above example flow is available here:

https://gist.github.com/ijokarumawak/6d2bbfe48809e036008153f15113ab80

[#3]

@kkawamura,

This simply worked PERFECT!

I am so grateful for your answer! You removed 999 tons of my back!

The only thing that I don't understand is why ExecuteSQL only understand the EvaluteJsonPath instead of a simply Avro or Json...

I saw the incoming data provenance in ExecuteSQL with the EvaluateJsonPath and removing EvaluateJsonPath processor and is the same!

Can you or someone explain which was the trick?

What this image below exactly do?

Than you so much @kkawamura!!!

[#4]

Hi @Gabriel Queiroz

Glad to hear that works! The reason I put EvaluateJsonPath before ExecuteSQL is, ExecuteSQL can user FlowFile attributes with NiFi Expression Language (EL).

If you haven't done yet, I recommend you to read Apache NiFi User Guide 'Terminology' [1] section. A FlowFile has two different data, Attributes and Content. Content is a opaque binary data, and Attributes is something like a hash map having string keys and values (META data).

Processor properties like ExecuteSQL 'SQL select query' support EL. By using EL, you can construct a string value (e.g. SQL) using FlowFile Attributes. I also recommend reading EL Guide [2].

On the other hand, usually the data itself such as JSON or Avro serialized data is stored as FlowFile content, which is not accessible from EL directly. So, I used EvaluateJsonPath to extract values from FlowFile content to FlowFile Attribute, so that ExecuteSQL can use it.

The user-defined properties at EvaluateJsonPath (in the red rectangle) are JSON paths. Both NiFi EL and JSON path use '$' so look similar but those are different techniques.

Does it answer to your question?

[1] https://nifi.apache.org/docs/nifi-docs/html/user-guide.html#terminology

[2] https://nifi.apache.org/docs/nifi-docs/html/expression-language-guide.html

[#5]

Thank you @kkawamura,

your answer was very clear!