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.
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.
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:
- https://blogs.apache.org/nifi/entry/record-oriented-data-with-nifi
- https://community.hortonworks.com/articles/138632/data-flow-enrichment-with-nifi-lookuprecord-proces...
I hope this is helpful.
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.
