How to parse/Extract attribute name and its value using NiFi

Question

so I have a file coming in following format:

{"Field_1":0.01,"abc_id":5,"pqr_id":"0","xyz_id":946715026,"count":130,"Epoch":"130723413","measurement":"Grad/s"} {"Field_2":0.01,"abc_id":2,"pqr_id":"5","xyz_id":841712547,"count":190,"Epoch":"130723414","measurement":"m/s2"}

-

.....and so forth I have multiples rows with attribute name: values.

What processor to use to separate the attribute as fields/columns and parse its respective values.

I want to parse the file in a way I can extract the field names and values separately as my ultimate goal is to feed the file data into MySQL(table schema in MySQL is defined Below:)

Field_1, Field_2, abc_id,pqr_id,xyz_id,count,Epoch,measurement

How can I achieve the whole use case in NiFi? I think I can achieve this using some kind of regular expression but I am not sure what regular expression will help me get the desired output. Any suggestion is much appreciated. Thanks!

user_9657108 · Answer

This can be done using ConvertJSONToSQL and PutSQL flow file processors. There are no additional steps required to parse JSON file.

ConvertJSONToSQL processor converts JSON formatted FlowFile into an UPDATE, INSERT, or DELETE SQL statement and does not directly insert data into Database. The incoming FlowFile is expected to be flat JSON message, meaning that it consists of a single JSON element and each field maps to a simple type. Upon successful conversion, the original FlowFile is routed to the 'original' relationship and the SQL is routed to the 'sql' relationship.

PutSQL executes a SQL UPDATE or INSERT command. The content of an incoming FlowFile is expected to be the SQL command to execute. This processor actually inserts the data into Database;

Your FlowFile:

{"Field_1":0.01,"abc_id":5,"pqr_id":"0","xyz_id":946715026,"count":130,"Epoch":"130723413","measurement":"Grad/s"}

Target table should be available in target database; Test setup uses database 'tempdb' and table 'test_db'.

Table DDL statement:

create table test_db (Field_1 double, abc_id int, pqr_id varchar(5), xyz_id int(11), count int, Epoch varchar(15), measurement varchar(30));

Also you need to create DBCPConnectionPool controller service which provides Database connection pooling service.

Attaching sample screenshots:

Flow:

DBCPConnectionPool:

Note that connection url has target database name.

ConvertJSONToSQL Processor:

GetFile processor could be any other processor(s)/flow that provides output flow file in JSON format.

How to parse/Extract attribute name and its value using NiFi

1 Answers