Today we will show how to interact with a NiFi instance to modify a flow at runtime via API.
Pre-requisites
- NiFi installed and running on a localhost: https://nifi.apache.org/download.html
- Groovy - because its JSON builders and REST DSLs are great. If you are on a Mac, the easiest is to run brew install groovy. To install Homebrew (the superb Mac package manager), visit http://brew.sh/
- Full script is available on GitHub: https://github.com/aperepel/nifi-rest-api-tutorial
- NiFi REST API Docs: https://nifi.apache.org/docs/nifi-docs/rest-api/index.html
Here’s the test flow we will be working with today:
Prepare the test flow:
- Add a PutFile component to the canvas.
- Rename the processor to Save File (right-click -> Configure -> Settings -> Name field). We will be using this name to look up the processor later via API.
- Add a GetHTTP processor and create a connection from GetHTTP to ‘Save File’. GetHTTP settings can be ignored for now, the Save File processor simply needs an input connected.
- Set the Save File properties as below (these settings will be modified programmatically next)
- Start the Save File processor (no need to start GetHTTP for our purposes)
Note: for a more complex flow, one would use templates: https://nifi.apache.org/docs/nifi-docs/html/user-guide.html#templates
Working with the API
Next, we will update the Save File processor to use a different directory (/tmp/staging) and set Create Missing Directories to true.
High-level script flow:
- Search the data flow for a component to operate on. Lookup term is 'Save File'. This is the same API used by the Search field in the UI.
- Validate there’s only 1 processor returned - we want to make sure we’re modifying the expected one.
- Sync up with the framework state - get the latest version field value, will be used in the update statement next. This is a classic Optimistic Locking pattern implementation.
- Build a small JSON document containing only state changes.
- Perform a partial update via a PUT operation.
- Repeat steps 4-5 to stop, update configuration (change directory and missing dirs properties) and start the processor.
For the impatient among us execute the script directly (clone/checkout the github if you want to play with the code later):
groovy https://raw.githubusercontent.com/aperepel/nifi-rest-api-tutorial/master/reconfigure.groovy
You will see an output similar to this:
Looking up a component to update...
Found the component, id/group: c35f1bb7-5add-427f-864a-bdd23bb4ac7f/f1a2c4e8-b106-4877-97d9-9dbca868fc16
Preparing to update the flow state...
Stopping the processor to apply changes...
Updating processor...
{
"revision": {
"clientId": "my awesome script",
"version": 309
},
"processor": {
"id": "c35f1bb7-5add-427f-864a-bdd23bb4ac7f",
"config": {
"properties": {
"Directory": "/tmp/staging",
"Create Missing Directories": "true"
}
}
}
}
Updated ok.
Bringing the updated processor back online...
OkIf you check the NiFi processor again, you will see the updated Directory and Create Missing Dirs. Additionally, every step has been captured and recorded in the flow history:
When you see a warning message in the UI, simply hit the Refresh link right next to it - I will explain the concurrency controls at the end of this article.
Code Walkthrough
First, we will pull in a dependency https://github.com/jgritman/httpbuilder/wiki/RESTClient . It is available in a public maven repository and is fetched automatically.
@Grab(group='org.codehaus.groovy.modules.http-builder',
module='http-builder',
version='0.7.1')
This allows us to use nice REST DSL like these:
nifi.get(
path: 'controller/search-results',
query: [q: processorName]
)
nifi.put(
path: "controller/process-groups/$processGroup/processors/$processorId",
body: builder.toPrettyString(),
requestContentType: JSON
)
Next, we are using Groovy's JSON builder to construct a JSON document for a partial PUT update, i.e. only specify the properties you want to change in the update, like this:
builder {
revision {
clientId 'my awesome script'
version resp.data.revision.version
}
processor {
id "$processorId"
config {
properties {
'Directory' '/tmp/staging'
'Create Missing Directories' 'true'
}
}
}
}
Those dot-notation variables navigate the JSON document tree from a previous response. To understand how to structure it, start by issuing a GET request against your processor, which will fetch a complete state document.
Tip: UI does everything through the REST API, it’s a great learning interactive learning tool in itself. One note, though, the UI will interchangeably leverage both PUT and POST (form) requests, so choose whichever is more convenient. In this write-up we will be using PUT with JSON.
Finally, the clientId and version business is explained in the next section.
Optimistic Locking in NiFi
The diagram below describes the concept.
Supplying a clientId is required for update operations to avoid running into consistency issues (the API will respond with 409 Conflict status code and it will be really confusing if a developer doesn’t know about this attribute).
controller/revision returns the clientId of a user who last modified the flow among other things. This is NOT always your id, best practice is to supply your own unique value to identify the client. It’s actually a free-form value, UUID is just a default that the framework generates for you if missing.
