nifi Json data using routeonattributeto filter attribute or attribute value

Question

I am currently working to consume data using Nifi to read tealium event stream and load into HDFS. Need help in filtering the data when source misses to send data for attribute.

{"account":"newtv","twitter:description":"Discover when your favorite NewTV shows and hosts are being shown. ","og:locale":"en_US","dcterms:publisher":"NewTV","original-source":"www.newtv.com/","og:url":"www.newtv.com/show/program-guide"}},"post_time":"2019-10-09 11:27:46","useragent":"Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/77.0.3865.90 Safari/537.36","event_id":"12345"}

Sample above message. I am currently stuck with filtering the data when source misses to send data for event_id attribute from the following sample dataset.

Current Nifi flow, Consume Kafka -> Evaluate Json Path -> Jolttransform Json -> Evaluate Json Path-> RouteOnAttribute -> Merge Content -> Evaluate Json Path -> Update attribute -> PutHDFS ->MoveHDFS

Need help how to split data using RouteOnAttribute to differentiate missing event_id attribute or attribute_value to two different flows. Flow with attribute or attribute value and missing values to error and load into different output path.

Answer 1

In EvaluateJsonPath processor add new property to extract event_id value from the flowfile.

if flowfile is not having event_id then nifi adds empty value to the attribute.

EvaluateJsonPath Configs:

Then by using RouteOnAttribute processor we can check the attribute value and route the flowfile accordingly.

RouteOnAttribute Configs:

not null value

${event_id:isEmpty():not()}

null value

${event_id:isEmpty()}

Then use null value and not null value relationships for further processing..!!