We’ve used NiFi on quite a few projects here at the labs. In each of those projects, there has been some form of a “reusable flow” component within. Many times, within these flows, we derive a target attribute using logic embedded in the Expression Language based on attributes attached to the FlowFile. This is useful when a downstream processor will utilize the attribute, like a target table, for example. Other times, we may simply be attaching a message based on the source attribute in a failure situation. The use cases for this design pattern are virtually endless.
One of the drawbacks of this pattern comes with the maintenance involved. Due to the nature of NiFi, any time a change must be made, or something needs to be added to the logic, this change may have to take place in many different processors. What follows is a very click-heavy and error-prone manual process. One way that we recently extended NiFi to improve this scenario is created a custom processor that uses an attribute to map to a corresponding key from a JSON object stored in the variable registry in NiFi.
If you’re familiar with NiFi you may be thinking, “isn’t this already available with some of the JSON processors that ship with NiFi?” Well, it is, but it makes things quite complicated. At the time that the extension was created, as far as I am aware, all of the processors that allowed something similar to be done, like EvaluateJSONPath, did so using the content of a FlowFile rather than a JSON object stored in the variable registry. Therefore, we would be complicating the flow quite a bit to enable the flow to leverage JSON from a variable. So, let’s just go over a really quick and dirty processor built to help keep things more organized and maintainable within NiFi.
The Snippet
First, let’s just take a look at the OnTrigger method of the processor.
If you’ve ever created a custom processor within NiFi, you’ve interacted with this method.
As you can see, there’s not a ton going on here. I’ll point out the 3 most important lines here.
- 74: Here we are getting the JSON Path, specified by the user, that will lead us to the relevant value associated with the key
- 75: We are creating a JSON Object from the JSON String stored in the variable registry
- 76: We are calling a method from the Util class that we have created to get the value stored in the JSON Path
It’s important to note that both the JSON Object and the JSON Path are specified in as a property associated with the Processor.
Now let’s look at the getJsonObjectFromPath method.
It’s understood that the jsonPath property is period (“.”) delimited. From there, it is a pretty simple process. We create an array of strings by splitting the path on each period. If there are no periods, we end up with an array of 1.
From there, we loop through the array and get the element associated with each key. If the path is invalid, we return null. Otherwise, we determine if the element is a JSON Object. If it is not, we return that element as a string. If it is a JSON Object, we replace the current JSON Object with the new JSON Object. Then we continue the loop until we reach the last element in the array, or the element is null or not a JSON Object.
Customizing Capabilities in NiFi
That’s it! Again, it is a quick and dirty implementation. There are definite enhancements that can be made. For example, we could move the assignment of the JSON Object property to the OnScheduled method if we wanted to cut down on the processing time to convert the string to a JSON object, if memory isn’t a concern.
Additionally, there is probably already a library out there, like json-path, that will allow you to feed it a path and a JSON Object and it will return the appropriate value. With that being said, we didn’t go down this path. We knew that we could add these enhancements later on if needed. What we wanted to point out most here is that capabilities that don’t ship with NiFi can easily be added with minimal code. Therefore, when an idea about a framework or capability comes to mind, and you feel that it might not be possible to implement it in NiFi, you’ll know that you aren’t pigeonholed to the capabilities that it ships with.
We use modern data technologies like NiFi to help our clients harness the power of their data. Learn more about how we approach our data engineering projects.
