question Re: Nifi : QueryDatabaseProcessor - Add new fields to Avro Files in Archives of Support Questions (Read Only)

Nifi : QueryDatabaseProcessor - Add new fields to Avro Files

ManishaJain — Fri, 03 Nov 2017 20:06:54 GMT

Have used QueryDatabase processor to read data from sql server DB.

It has read a avro files ,need to add static fields to the avro record

for ex..

=============================

Avro Record = id,name

Value= 1,ABC

=============================

need to add type to the record

========================================

New Avro Record = id,name,type

Value= 1,ABC,IN

========================================

Is there a way to manipulate avro files .

Re: Nifi : QueryDatabaseProcessor - Add new fields to Avro Files

Shu_ashu — Sun, 18 Aug 2019 07:49:01 GMT

@manisha jain

Yeah you can manipulate results coming out from your Query database processor itself.

if you want to add static fields to the each avro record then in your Querydatabase processor.

change Columns to return property as

*,'in' type //return all columns from table and add type column with value in for each row.

Processor Config:-

as you can view above screenshot how we are configuring Columns to Return property above.

(or)

Another way doing this is

Once you get results from Query database processor then

1.Use convertavrotojson processor //we are converting avro data into json.
2.Use ReplaceText processor //to search for literal value } and do replacement value as your desired value in jsonformat
ex:- ,"type":"in"}
in this processor using replace text we are having json array of records and we are looking for } that means at end of each record and we are going to replace with ,"type":"in"} for each record so the json record structure will wont change.
3.Use convertjsontoavro processor //we are converting json data into avro data again and this data having new field that we have added to the json message.

Re: Nifi : QueryDatabaseProcessor - Add new fields to Avro Files

ahadjidj — Sun, 18 Aug 2019 07:48:54 GMT

Hi @manisha jain

The approach described above (Avro -> Json -> Avro) is no more required with new record based processors in NiFi. You can use UpdateRecord processor to add new attribute to your flow files whatever their format is (Avro, JSON, CSV, etc).

Just define your Avro Schema with the new attribute and use in Avro Writter. Then use UpdateAttribute to add the value as in the below example where I add city attribute with the static value Paris:

And the result will look like

Note that in my example data is in JSON format but the same approach works for Avro. Just use Avro reader/writer instead of Json.

If you are new to record based processors read these two articles:

I hope this is helpful.

Re: Nifi : QueryDatabaseProcessor - Add new fields to Avro Files

ManishaJain — Tue, 07 Nov 2017 17:18:55 GMT

Thanks @Shu

Will try this approach and revert.

Re: Nifi : QueryDatabaseProcessor - Add new fields to Avro Files

ManishaJain — Tue, 07 Nov 2017 17:20:06 GMT

Thanks @Abdelkrim Hadjidj . Very informative .

Re: Nifi : QueryDatabaseProcessor - Add new fields to Avro Files

abhinav_joshi — Mon, 26 Mar 2018 10:05:17 GMT

Great Answers thanks all

Re: Nifi : QueryDatabaseProcessor - Add new fields to Avro Files

abhinav_joshi — Thu, 26 Apr 2018 08:34:50 GMT

@Shu

Hey I have a requirement to add a new column to the extracted contents of an oracle table. The new column name is LOAD_TMS and should contain the current timestamp. I used your approach and typed the following in the "columns to return" property - *,'${now():toNumber():format('yyyy-MM-dd HH:mm:ss')}' LOAD_TMS however i am getting errors . I also tried copying your example but even then I started getting irrelevant errors like FROM key word is missing.

Can you please assist here

Thanks

Abhi

Re: Nifi : QueryDatabaseProcessor - Add new fields to Avro Files

abhinav_joshi — Thu, 26 Apr 2018 12:09:47 GMT

@Shu

Meanwhile , I tried using your other option and that worked which is convert Avro to JSON and then using a replace text. That was a cool approach 🙂 . However , we are doing a streaming ingestion and want to limit the number of processors , and hence prefer option 1 which is putting that extra column at the "Query Database Table" processor layer itself.

cheers

Abhi

Re: Nifi : QueryDatabaseProcessor - Add new fields to Avro Files

Shu_ashu — Sun, 18 Aug 2019 07:48:40 GMT

@Abhinav Joshi

Which version of NiFi are you using?
In NiFi 1.1 Columns to Return property doesn't support expression language probably that's the reason why you are having issues while adding '${now():toNumber():format('yyyy-MM-dd HH:mm:ss')}' LOAD_TMS this expression.

If your columns to return doesn't support expression language then one of the possible work around is convert to json then add field again convert back to avro.

In New versions of i think from NiFi 1.2+ Columns to Return property supports expression language and also we are having update record processor by using this processor we can add new fields to the flowfile content without converting them.

Re: Nifi : QueryDatabaseProcessor - Add new fields to Avro Files

abhinav_joshi — Thu, 26 Apr 2018 14:25:13 GMT

Hey thanks . I am using 1.4 version but still I am not able to get this working. Please find the screen shot below columns-from-rdbms.png

So , I can see that the regular expression is working as in the error section , the timestamp is coming up however the issue perhaps seems to be with the syntax which is *, ............... . Isnt this supported ? Waiting for your brilliant feedback as always @Shu

Re: Nifi : QueryDatabaseProcessor - Add new fields to Avro Files

abhinav_joshi — Fri, 27 Apr 2018 07:00:51 GMT

Thanks @Shu for trying this out . I used the double quotes but it isnt working with Oracle. For now , I will use your JSON approach as a stop gap arrangement till I find a way out to fix this. In case you have any ideas then do let me know. Thansk again

Re: Nifi : QueryDatabaseProcessor - Add new fields to Avro Files

abhinav_joshi — Fri, 27 Apr 2018 10:06:35 GMT

Ok got this fixed . In oracle , this works well if we add the table name and then a .*. So this is what I put in and my problem got fixed

GIM.GIDB_GC_SKILL.*,'${now():toNumber():format('yyyy-MM-dd HH:mm:ss')}' AS LOAD_TMS

Thanks

Re: Nifi : QueryDatabaseProcessor - Add new fields to Avro Files

abhinav_joshi — Fri, 27 Apr 2018 15:40:15 GMT

@Ok , I managed to get this fixed. In oracle I can see it works a little differently. I used the following and this worked.I passed the table name followed by a "." and that seemed to do the trick. @Shu

GIM.GIDB_GC_SKILL.*,'${now():toNumber():format('yyyy-MM-dd HH:mm:ss')}' AS LOAD_TMS