Overview of Controller Services

In Apache NiFi, Controller Services are not the same as the controllers found in MVC patterns. They are standalone configurable services that operate at a similar level to processors, but instead of processing data directly, they provide shared resources or capabilities that processors can leverage. For example, a JDBC connection pool service provides database connectivity for SQL-based processors.

Controller Services are extensible, allowing custom implementations to be developed just like custom processors.

When a processor requires a service, its configuration UI presents options to select an existing service or create a new one. Creating a new service generates a "global" service object scoped to the current process group, which can then be reused by other processors requiring the same service type.

Configuring Controller Services

To view and manage Controller Services, right-click on an empty area of the canvas and select Configure. A list of services appears, showing name, type, description, enabled status, and actions.

Before modifying a service's properties, it must be disabled. Click the configure icon to open the properties dialog, which has three tabs similar to processor configuration: SETTING, PROPERTIES, and COMMENT.

SETTING Tab

This tab contains the service name (which can be changed) and a list of processors that reference this service.

PROPERTIES Tab

The properties tab holds the core business configuration. The fields vary depending on the service type. For a DBCPConnectionPool, common properties include:

• Database connection URL
• Database username and password
• Database driver class name
• Path to the driver JAR file(s)

Many services require third-party JAR dependencies. For production, its recommended to store all required JAR files in a centralized location.

COMMENT Tab

This tab provides free-form documentation for the service. Users can add usage notes or descriptions as needed.

Scope of Controller Services

Process groups affect the scope of Controller Services. A service created at the root level (outside any process group) is accessible throughout the entire NiFi instance. When created inside a specific process group, it can only be referenced by processors within that group. This scoping mechanism simplifies service management and reuse.

Global Parameter Contexts

When many instances of similar services (e.g., database connection pools, Kafka clients, Redis clients) are created, repetitive configuration of URLs, usernames, and passwords becomes tedious. To address this, NiFi supports Parameter Contexts.

Before using variables: URLs, driver class names, database users, etc., must be entered manually for each service, increasing maintenance overhead.

To simplify, create a Parameter Context:

1. Open the Parameter Context configuration.
2. Define a set of parameters (e.g., db_url, db_user, db_driver).
3. Apply the context to the appropriate process group(s).

Then, when configuring a service's properties, replace literal values with parameter references using the # prefix:

Database Connection URL: #{db_url}
Database User: #{db_user}
Database Driver Class Name: #{db_driver}

This centralizes configuration and eases maintenance.

Example: DBCPConnectionPool with ExecuteSQL

The following example demonstrates reading data from MySQL, transforming it, and outputting the results.

Step 1: Create a DBCPConnectionPool Service

• In the Controller Services panel, create a new DBCPConnectionPool.
• Configure the required properties (URL, driver class, credentials, etc.).
• Enable the service.

Step 2: Add ExecuteSQL Processsor

• Add an ExecuteSQL processor to the canvas.
• Configure its properties, selecting the DBCPConnectionPool service created above.
• Write a SQL query to read data from MySQL.

Step 3: Convert AVRO to JSON

• Add a ConvertAvroToJSON processor after ExecuteSQL.
• Connect the output of ExecuteSQL to this processor.
• This step makes the data human-readable and easier to process in subsequent steps.

Step 4: Process Data with ExecuteGroovyScript

• Add an ExecuteGroovyScript processor.
• Connect the JSON output to this processor.
• Use a Groovy script to filter or transform the data. Example script:

import org.apache.commons.io.IOUtils;
import java.nio.charset.StandardCharsets;
import groovy.json.JsonSlurper;
import groovy.json.JsonOutput;

def inputData = getInputJSONData()
if (inputData == null) return

def results = []
inputData.each { item ->
    // Perform data processing here
    results.add(item.name)
}

if (results.size() > 0) {
    sendOutput(results, REL_SUCCESS)
}

def getInputJSONData() {
    def flowFile = session.get()
    if (flowFile == null) return null
    
    def jsonStr = ''
    session.read(flowFile, { inputStream ->
        jsonStr = IOUtils.toString(inputStream, StandardCharsets.UTF_8)
    } as InputStreamCallback)
    
    def data = null
    try {
        data = new JsonSlurper().parseText(jsonStr)
    } catch (Exception e) {
        log.error('Invalid input JSON')
    }
    session.remove(flowFile)
    return data
}

def sendOutput(data, relationship) {
    def jsonString = JsonOutput.toJson(data)
    def newFlowFile = session.create()
    newFlowFile = session.write(newFlowFile, { outputStream ->
        outputStream.write(jsonString.getBytes(StandardCharsets.UTF_8))
    } as OutputStreamCallback)
    session.transfer(newFlowFile, relationship)
}

Step 5: Log or Save Results

• Add a LogMessage processor (or any desired sink) to receive the processed data.

Using Controller Services Inside ExecuteGroovyScript

ExecuteGroovyScript can access Controller Services from within the script. This is useful when the script needs to query a database or use other services.

Setup

1. In the ExecuteGroovyScript processor's PROPERTIES tab, add a new property.
2. When naming the property, prefix it with SQL. or CTL. to enable service binding.
3. Select the desired service from the dropdown that appears after using the prefix.

Script Example

The script can then use the service via SQL.<service_name>.<method>:

import org.apache.commons.io.IOUtils;
import java.nio.charset.StandardCharsets;
import groovy.json.JsonSlurper;
import groovy.json.JsonOutput;

def inputData = getInputJSONData()
if (inputData == null) return

def results = []
inputData.each { item ->
    def lookupMap = [:]
    // Query database using service
    SQL.mysql.eachRow('SELECT id, value FROM tb_dic_detail WHERE u_status = 1') { row ->
        lookupMap.put(row.id.toString(), row.value.toString())
    }
    
    // Enrich or process data
    results.add(item.name)
}

if (results.size() > 0) {
    sendOutput(results, REL_SUCCESS)
}

def getInputJSONData() {
    def flowFile = session.get()
    if (flowFile == null) return null
    
    def jsonStr = ''
    session.read(flowFile, { inputStream ->
        jsonStr = IOUtils.toString(inputStream, StandardCharsets.UTF_8)
    } as InputStreamCallback)
    
    def data = null
    try {
        data = new JsonSlurper().parseText(jsonStr)
    } catch (Exception e) {
        log.error('Invalid input JSON')
    }
    session.remove(flowFile)
    return data
}

def sendOutput(data, relationship) {
    def jsonString = JsonOutput.toJson(data)
    def newFlowFile = session.create()
    newFlowFile = session.write(newFlowFile, { outputStream ->
        outputStream.write(jsonString.getBytes(StandardCharsets.UTF_8))
    } as OutputStreamCallback)
    session.transfer(newFlowFile, relationship)
}

This approach enables powerful data processing flows that combine NiFi processors with custom scripting logic.