trying to add a field based on conditionals with NifI

user_16538381 0
Viewed 186

I'm new to Apache NiFi and currently using it to route message data to various locations. I'm looking to add some fields based on a set of conditionals.

Currently I have a GetFile Processor that reads log files ---> ExtractGrok that applies a grok pattern to parse ---> ConvertRecord to convert from Grok to Json. The next part is where I'm stumped/not sure what to do next.

In my json I have a field refresh_time I need to create 2 new fields based on some conditions about the field refresh_time

something along the lines of if refresh_time < 10 then cache = 1; else if refresh_time > 10 then reprocess = 1

The end goal here is numeric fields cache and refresh_time that can be used down the road in aggregations.

What would be the best way to add 2 numerical fields based on a condition. Is there a processor for adding additional fields or updating the record to include new fields?

Thanks.

apache-nifi
1 Answers

There's a couple ways you could achieve what you want to.

One option (More readable)

A QueryRecord would let you write a SQL statement across your Records and let you split them by the result. E.g.

Add a dynamic property called cache with a value SELECT * FROM FLOWFILE WHERE refresh_time < 10.

Add a dynamic property called refresh with a value SELECT * FROM FLOWFILE WHERE refresh_time > 10.

The QueryRecord will now have the relationships failure, original, cache and refresh.

Branching off from cache and refresh will be one UpdateRecord each, with Replacement Value Strategy set to Literal Value.

For the cache relationship, you can add a new dynamic property called cache with a value 1. For the refresh relationship, you can add a new dynamic property called refresh with a value 1.

Similar option (Possibly more performant)

If you want to avoid the additional UpdateRecord, you can add fields in the QueryRecord with something like this:

Two dynamic properties set as:

cache = SELECT *, 1 AS cache FROM FLOWFILE WHERE REFRESH < 10

reprocess = SELECT *, 1 AS reprocess FROM FLOWFILE WHERE REFRESH > 10

This option may be more performant due to fewer disk reads.

This gist is an example of the second option, you can import it to NiFi to try it out.

Also, FYI there is a GrokReader that you could use in ConvertRecord to parse with Grok straight to JSON, potentially skipping the ExtractGrok.

user_39559720
Related