Forem: Jannik Rebmann

Workbench for Apache NiFi data flows

Jannik Rebmann — Fri, 17 May 2024 09:20:58 +0000

This article presents the concept and implementation of a universal workbench for Apache NiFi data flows.

The workbench is intended to improve the quality of your data flows by increasing their:

Testability - How to test the specified functionality of my data flows?
Extensibility - How to ensure the functionality of my data flows after changes?
Reliability - How to define and test edge cases of my data flows?

The article is primarily aimed at advanced Apache NiFi users, but is also of interest to beginners who are in the process of learning basic development concepts.

Motivation

Following up my last post Setup a secure Apache NiFi cluster in Kubernetes, I would now like to cover an important topic regarding the quality of the Apache NiFi data flow.

Before I began using Apache NiFi, I took a close look at its tools and concepts. While I was excited about its potential, I found the standards and tools for developing data flows lacking. To be honest, I was expecting something like this:

The green lights give me a feeling of security and well-being. It signals to me that I can make adjustments and changes to the code without having to worry about destroying the functionality. Also, unit tests give me an entry point to understand programs and develop new features.

Even if it is a bit challenging at first, test-driven development is a very stable approach and prevents problems before they occur. Especially when extensions and refactorings are required after the first releases, the test-driven approach is unbeatable in my eyes.

That doesn't mean that there aren't other great approaches. But unfortunately, I couldn't find any established approaches or best practices for developing data flows in Apache NiFi. Apart from a few articles on unit testing custom processors and using GenerateFlowFile processors for basic smoke testing, the number of articles about securing the functionality of data flows is small.

Working in an environment that's always changing, where adapting functionality and ensuring reliability are key, I realized I needed to develop my own structured approach.

That's when I came up with the idea of a Data Flow Workbench.

The basic idea is straightforward:
Simply encapsulate a data flow within a predefined and immutable context. Inputs and expected outputs remain the same, while only the data flow itself changes. This approach is similar to the principles of unit testing but needs to be tailored for data flow scenarios.

Concept

Lets dive deeper into the concept of the Apache NiFi Workbench.

Apache NiFi data flows can be organized within process groups, which is similar to organize code within functions. A ProcessGroup has defined inputs and outputs (represented by input and output ports) that are connected by processing pipelines of process groups and processors. A FlowFile is the data which flows through the piplines and consists of attributes and a content.

Now you know everything you need to know about data flows in order to understand the following illustration, which shows the basic structure of the Workbench. The process group PG main contains the data flow to be tested.

Splitting the flow files into their attributes and contents as separate inputs allows them to be defined as separate files. The workbench requires the attributes as JSON, while the content can be in any file format. The flow files of all inputs are related via the assert.group.identifier attribute. This ensures that the correct parts are put together.

An example input can look like this:

	`[input]` Attributes	`[input]` Content	`[input]` Expected Attributes	`[input]` Expected Content
`[attribute]` `assert.group.identifier`	test	test	test	test
`[content]`	`{ "done": false }`	Hello	`{ "done": true }`	Hello world

Workflow

As everywhere in software engineering, the quality of the software depends heavily on a proper design. For this reason, a clear and modular structure of the data flows is strongly recommended. In particular, it makes sense to decouple the data sources and data targets from the processing logic.

---------------      -----------     ---------------
| Data Source |  --> | PG_main | --> | Data Target |
---------------      -----------     ---------------

If that's the scenario, the processing logic PG_main can be integrated into the workbench as an isolated module, allowing us to define simulated inputs along with their corresponding expected outputs.

However, the workbench truly unleashes its full potential when paired with the Apache NiFi Registry. Once changes are made to PG_main, they can be committed to version management and seamlessly integrated into productive data flows.

Implementation

TL;DR In this section the implementation of the workbench is explained in more detail.
If you like to test and explorer the workbench yourself, you can directly jump to the conclusion to download the workbench template.

The workbench consists of two core modules (implemented as process groups):

Build FlowFile: Builds a FlowFile from given attributes and a specific content.
Assert FlowFile: Compares the attributes and the content of two flow files and fails if they are different or pass if they are equal.

The implementation of the two core modules is described in more detail in the following two sections.

Process Group: `Build FlowFile`

This module merges the attributes with their corresponding content into a single FlowFile. Sounds simple for common programming languages, but is tricky for data flows, as there is no defined order in which files arrive.

NiFi provides a standard processor (MergeContent) for this purpose. The Create fragments processors ensure that the required attributes of the MergeContent processor are properly configured for incoming flow files. After the merge, the mime type is restored and some helper attributes are cleaned up.

Everything is actually pretty straight forward, but you may have noticed that I skipped the JSON to attributes processor. Let's take a closer look. Unfortunately there is no standard processor for extracting a JSON from the flow file content as attributes. So it needs to be implemented.
This can be done with the ExecuteScript processor, which makes it possible to execute user-defined code on incoming flow files. There are various programming languages to choose from and I decided to use Jython (python running on the Java plattform).

The following code retrieves a flow file, reads and parses its content as JSON and adds the keys with their values as attributes.

import json
from org.apache.commons.io import IOUtils
from java.nio.charset import StandardCharsets
from org.apache.nifi.processor.io import InputStreamCallback

class JsonInputStreamCallback(InputStreamCallback):
    def __init__(self):
        pass
    def process(self, inputStream):
        jsonStr = IOUtils.toString(inputStream, StandardCharsets.UTF_8)
        self.dict = json.loads(jsonStr)

flowFile = session.get()
if flowFile != None:
    try:
        callback = JsonInputStreamCallback()
        session.read(flowFile, callback)
        newFlowFile = session.create(flowFile)
        newFlowFile = session.putAllAttributes(newFlowFile, callback.dict)
        session.remove(flowFile)
        session.transfer(newFlowFile, REL_SUCCESS)
    except Exception as e:
        log.error("An error occured", e)
        session.transfer(flowFile, REL_FAILURE)

Process Group: Assert FlowFile

The second core module compares the actual flow file against the expected one. The subsequent flow expands on the idea by adding the hash of the file's content as an attribute, and then proceeds to compare all existing attributes.
There are some core attributes like uuid and filename (see CoreAttributes for the complete list) that are ignored in the comparison.

The heart of the data flow is the processor Assert FlowFile which compares two flow files with each other. As Apache NiFi does not provide a standard processor for this either, this logic is also implemented using an ExecuteScript processor.

It tries to find the corresponding actual or expected flow file to be compared. If there is no matching file with the same attribute assert.group.identifier it is pushed back to the queue. Otherwise the attributes of both files get compared and asserted.

from java.util import HashMap
from org.apache.nifi.flowfile.attributes import CoreAttributes
from org.apache.nifi.processor import FlowFileFilter
from org.apache.nifi.processor.FlowFileFilter import FlowFileFilterResult

FILE_TYPE_KEY = 'assert.file.type'
GROUP_IDENTIFIER_KEY = 'assert.group.identifier'

class FlowFileTypeFilter(FlowFileFilter):
    def __init__(self, type, identifier = None):
        self._type = type
        self._identifier = identifier
    def filter(self, flowFile):
        type = flowFile.getAttribute(FILE_TYPE_KEY)
        if type == self._type:
            if self._identifier:
                identifier = flowFile.getAttribute(GROUP_IDENTIFIER_KEY)
                if identifier == self._identifier:
                    return FlowFileFilterResult.ACCEPT_AND_TERMINATE
                else:
                    return FlowFileFilterResult.REJECT_AND_CONTINUE
            else:
                return FlowFileFilterResult.ACCEPT_AND_TERMINATE
        else:
            return FlowFileFilterResult.REJECT_AND_CONTINUE

def getFlowFile(type, identifier = None):
    flowFiles = session.get(FlowFileTypeFilter(type, identifier))
    if not flowFiles.isEmpty():
        return flowFiles[0]
    else:
        return None

def getAttributes(flowFile):
    map = HashMap(flowFile.getAttributes())
    map.remove(FILE_TYPE_KEY)
    for attr in CoreAttributes.values():
        attrName = attr.key()
        if map.containsKey(attrName):
            map.remove(attrName)
    return map

def compare(actual, expected):
    actualAttrs = getAttributes(actual)
    expectedAttrs = getAttributes(expected)
    if len(actualAttrs) != len(expectedAttrs):
        return "The number of attributes differs"
    for key in actualAttrs:
        if expectedAttrs[key] != actualAttrs[key]:
            return 'Attribute "{}" differs (actual / expected): {} / {}'.format(
                key, actualAttrs[key], expectedAttrs[key])
    return None

actual = getFlowFile("actual")
if actual != None:
    expected = getFlowFile(
        "expected", actual.getAttribute(GROUP_IDENTIFIER_KEY))
    if expected != None:
        try:
            diffMsg = compare(actual, expected)
            session.remove(expected)
            actual = session.removeAttribute(actual, FILE_TYPE_KEY)
            if diffMsg:
                actual = session.putAttribute(actual, 'assert_message', diffMsg)
                session.transfer(actual, REL_FAILURE)
            else:
                session.transfer(actual, REL_SUCCESS)
        except Exception as e:
            log.error('Something went wrong', e)
            session.rollback(True)
    else:
        session.rollback(True)

Conclusion

The workbench presented in this article can be used universally for all isolated dataflows in Apache NiFi. It therefore offers a great extension for the development and testing of data flows and thus makes a valuable contribution to increasing the quality of data flows.

The attributes and content of the flow files can be input separately in the Workbench. This characteristic makes the concept easily extendable to a test suite for Apache NiFi, which I will present in a follow-up article.

If you want to look at the workbench in detail feel free to download the workbench as an Apache NiFi 1.25 template:

Download Workbench

Setup a secure Apache NiFi cluster in Kubernetes

Jannik Rebmann — Thu, 23 Nov 2023 19:27:52 +0000

This article provides a detailed, step-by-step guide on setting up a secure Apache NiFi cluster with a NiFi Registry in Kubernetes, featuring the following capabilities:

NiFi and the NiFi Registry are secured via https.
Authentication of all services is realized via OpenId Connect (OIDC).
The internal communication between the nodes is encrypted.
The communication between the cluster and the NiFi Registry is encrypted and authenticated.

Motivation

In a world where ChatGPT is bringing artificial intelligence into our everyday lives, data integration becomes a key challenge. Ensuring that AI systems receive the right data at the right time will be crucial.

This article addresses this challenge using Apache NiFi, a proven data integration system that has been effectively solving data integration problems long before the AI revolution.

However, I do have one major criticism of Apache NiFi: The barrier to entry is relatively high. The process of setting up a secure cluster of nodes and connecting it to a secure NiFi registry can be time-consuming, especially for those new to the system.

That is why I have decided to write this article to help everyone get started with this excellent system. I promise it will be worth it!

Prerequisites

You need a Linux system (I used Ubuntu 22.04.3 LTS) with the following software installed:

docker: Platform and runtime environment for container virtualization.
minikube: A local Kubernetes environment for development.
helm: A package manager for organizing software and systems developed for Kubernetes.

1. Preparations

The use of minikube is very simple. With just one command, a local Kubernetes is started, which offers all the features of Kubernetes except for the high scalability:

$> minikube config set cpus 4

$> minikube config set memory 8184

$> minikube start

1.1 Enable ingress in minikube

To be able to access the Apache NiFi services via URL later on, we still need to enable the ingress controller of minikube.

$> minikube addons enable ingress

1.2 Integrate `kubectl` for minikube

Another useful thing about minikube is that it always comes with a matching kubectl client. This can be accessed with the command minikube kubectl and behaves identically to a standalone installation of kubectl. Therefore it is recommended to provide this command with an alias and enable auto-completion for kubectl.

$> echo 'alias kubectl="minikube kubectl --"' >> ~/.bashrc

$> echo 'source <(kubectl completion bash)' >> ~/.bashrc

$> source ~/.bashrc

After all the above steps are done, everything is set up to use kubectl against minikube. To test kubectl, you can run the following command (you can use auto-completion with Tab) :

$> kubectl version -o yaml

1.3 Install cert-manager in minikube

The cert-manager is a framework to organize X.509 certificates within a Kubernetes cluster and simplifies the process of obtaining, renewing and using certificates.

To secure the NiFi cluster from unauthorized access and to encrypt the communication between the NiFi nodes, we use the cert-manager to issue us these certificates.

The installation of the cert-manager can be done with a single command:

$> kubectl apply -f https://github.com/cert-manager/cert-manager/releases/download/v1.12.0/cert-manager.yaml

1.4 Map NiFi domains to minikube

While the ingress controller handles URL mapping within the Kubernetes cluster, it's important to note that the URL must initially reach minikube. After successfully configuring the setup, you will be able to access the following two addresses via your browser:

nifi.example.org
nifi-registry.example.org

You can use the following command to add both mappings to the /etc/hosts file:

$> cat << EOF | sudo tee -a /etc/hosts
# Map nifi.example.org and nifi-registry.example.org to minikube ip
`minikube ip`   nifi.example.org
`minikube ip`   nifi-registry.example.org
EOF

1.5 Register OpenID connect (OIDC) clients

OpenID Connect (OIDC) is a protocol for secure user authentication and information sharing, where the provider performs authentication on behalf of the application. OIDC has become a standard and is offered by many large platform providers such as Google, PayPal but also GitLab.
Moreover, OIDC is gaining popularity within organizations. Solutions such as Keycloak or Authelia offer a convenient ways to provide OpenId Connect on the basis of e.g. LDAP.

Both Apache NiFi and the Apache NiFi Registry support OpenID Connect to authenticate their users. This means we have to register two clients with a OIDC provider.

For this article I will use GitLab as OIDC provider. However, any other platform can be used as well. The only thing that changes is actually the domain name. The required information remains the same.

For GitLab the OIDC client registration is very easy. Just open your GitLab profile and create two new Applications:

NiFi OIDC Client:

Name: NiFi
Redirect URI: https://nifi.example.org/nifi-api/access/oidc/callback
Scopes:
- openid
- email

NiFi Registry OIDC Client:

Name: NiFi Registry
Redirect URI: https://nifi-registry.example.org/nifi-registry-api/access/oidc/callback
Scopes:
- openid
- email

For both registrations you need to save the following information for later integration into our NiFi services:

Name	Placeholder	Example
Discovery URL	`<discovery_url>`	https://gitlab.com/.well-known/openid-configuration
Application ID	`<application_id>`	c9515c774fa1036cbcae5de455a23cc6ca7da54109a858f5b2c6869a89d40f08
Secret	`<secret>`	90b45e16b759fa097461917e7ef3df2c79916b548e1dc44f8d1c2b2c8a8c5537
GitLab email	`<registered_email>`	john.doe@example.org

"The OIDC standard uses a discovery endpoint (discovery_url) to supply clients with configuration information from the OIDC server. This endpoint URL consistently ends with .well-known/openid-configuration but might have a unique prefix path depending on the provider.

For instance, Keycloak includes additional realm information in its discovery URL: https://{keycloakhost}:{keycloakport}/realms/{realm}/.well-known/openid-configuration.

2. Setup an Apache NiFi cluster

One major advantage of Kubernetes standardization is the availability of numerous preconfigured software packages, including entire software systems in the form of helm packages. Fortunately, there's a helm chart for Apache NiFi, simplifying the process of setting up a entire cluster.

You can access this package through a public helm repository, which you can conveniently add to your local helm chart sources using these commands:

$> helm repo add cetic https://cetic.github.io/helm-charts

$> helm repo update

This helm chart provides a wide range of configuration options, all documented in the associated GitHub project helm-nifi.

The following configuration deploys a secure NiFi cluster with two nodes and OIDC authentication (the placeholders <application_id>, <secret> and <registered_email> are defined in section 1.5):

fullnameOverride: nifi
image:
  tag: 1.18.0
replicaCount: 2
properties:
  sensitiveKey: changeMechangeMe
  isNode: true
  webProxyHost: nifi.example.org
certManager:
  enabled: true
  ## Uncomment the next two lines only if you have
  ## installed the NiFi registry
  # caSecrets:
  #   - nifi-registry-ca
auth:
  admin: <registered_email>
  oidc:
    enabled: true
    discoveryUrl: https://gitlab.com/.well-known/openid-configuration
    clientId: <application_id>
    clientSecret: <secret>
    claimIdentifyingUser: email
    admin: <registered_email>
persistence:
  enabled: true
  subPath:
    enabled: true
ingress:
  enabled: true
  className: nginx
  annotations:
    nginx.ingress.kubernetes.io/app-root: /nifi
    nginx.ingress.kubernetes.io/backend-protocol: HTTPS
    nginx.ingress.kubernetes.io/affinity-mode: persistent
    nginx.ingress.kubernetes.io/affinity: "cookie"
    cert-manager.io/issuer: "nifi-ca"
  hosts:
    - nifi.example.org
  tls:
    - hosts:
        - nifi.example.org
      secretName: nifi-example-crt-secret

Now you can deploy the Apache NiFi cluster with the given configuration file nifi_values.yaml:

$> helm upgrade -i -f nifi_values.yaml nifi cetic/nifi

Open your browser and enter the address https://nifi.example.org.

Depending on your PC and internet connection, the download of all Docker images and the startup of the entire system can take several minutes.

You can check the current progress by entering:
$> kubectl get pods
If all pods are in the READY state, you should be able to access the service via the browser.

You may have to accept your browser's certificate warning.

3. Setup an Apache NiFi Registry

As with the installation of the NiFi cluster, there is a helm package for the NiFi registry. You need to add it to your local helm repository:

$> helm repo add dysnix https://dysnix.github.io/charts/

$> helm repo update

The following configuration deploys a secure NiFi registry with OIDC authentication (the placeholders <application_id>, <secret> and <registered_email> are defined in section 1.5):

fullnameOverride: nifi-registry
image:
  tag: 1.18.0
security:
  enabled: true
  needClientAuth: false
  admin: <registered_email>
certManager:
  enabled: true
  replaceDefaultTrustStore: false
  caSecrets:
    - nifi-ca
  additionalDnsNames:
    - nifi-registry
oidc:
  enabled: true
  discoveryUrl: https://gitlab.com/.well-known/openid-configuration
  clientId: <application_id>
  clientSecret: <secret>
  claimIdentifyingUser: email
  admin: <registered_email>
persistence:
  enabled: true
ingress:
  enabled: true
  className: nginx
  annotations:
    nginx.ingress.kubernetes.io/app-root: /nifi-registry
    nginx.ingress.kubernetes.io/backend-protocol: HTTPS
    cert-manager.io/issuer: " nifi-registry-ca"
  hosts:
    - host: nifi-registry.example.org
      paths:
        - path: /
          pathType: Prefix
  tls:
    - hosts:
        - nifi-registry.example.org
      secretName: nifi-reistry-example-crt-secret

Now you can deploy the Apache NiFi Registry with the given configuration file nifi_reg_values.yaml:

$> helm upgrade -i -f nifi_reg_values.yaml nifi-reg dysnix/nifi-registry

Open your browser and enter the address https://nifi-registry.example.org.

4. Configure Apache NiFi

Since we have enabled authentication with the NiFi Registry, the NiFi cluster must also authenticate itself to the registry. For this to happen, both services need to trust each other.

To do this, we have to uncomment the caSecrets configuration block from the nifi_values.yaml. This imports the newly available NiFi Registry certificate into the local truststore of the NiFi nodes. To activate the change, you need to deploy configuration again. If you have copied the whole nifi_values.yaml originally you can use following command to uncomment the lines and redeploy the NiFi nodes:

$> sed -i 's/#[^#]//' nifi_values.yaml && helm upgrade -i -f nifi_values.yaml nifi cetic/nifi

After the cluster nodes have been rebooted, you will need to add a new NiFi registry to the NiFi settings. Navigate to the burger menu at the top right of the NiFi UI and opening the "Controller Settings" menu. Go to the "REGISTER CLIENTS" tab and register a new client using the "+" symbol.
Give it a name and an optional description and save it.
You will then need to edit the newly created entry, go to the "PROPERTIES" tab and set the URL to https://nifi-registry:18080.

This URL https://nifi-registry:18080 is the internal service address within the cluster. Do not use the external URL https://nifi-registry.example.org otherwise the node authentication will fail.

Now only the node authentication is left. For this purpose, each NiFi node is created as a separate user. These can be found by clicking on the burger menu at the top right of the NiFi UI and opening the "Users" menu.

Exactly these users must now be created and authorized in the NiFi Registry. To do this, go to the NiFi Registry UI and click on "LOGIN" in the upper right corner.

Once GitLab has successfully authenticated you, you will be redirected back to your NiFi registry and your username should show up along with a "Settings" icon.

You can click it and create a new "Bucket".

Then switch to the "USERS" Tab and create the NiFi node users with read permissions on "Can manage buckets" and read, write and delete permissions on "Can proxy user requests".

If all steps have been completed successfully, you should now see the "Sample Bucket" in the NiFi UI when you import a "Process Group" from the Registry (drag&drop a Process Group from the header menu and click Import from Registry).

4. Limitations and Troubleshooting

The helm charts cetic/nifi (1.1.4) and dysnix/nifi-registry (1.1.4) are very helpful, but have some limitations.

The latest working NiFi and NiFi Registry docker versions are 1.18.0. The following container versions use Java 11, which brings a new default format (PKCS12 instead of JKS) for the truststore (breaking change).
When deploying the NiFi Registry, a permission error may occur on the auth-conf folder. This is due to a bug in helm chart which omits the setting of permissions on this folder. Either the permissions on this folder have to be set manually once, or the folder has to be added to the initContainers script.
Due to the use of the cert-manager, certificates are issued on the fully qualified internal Kubernetes service name. This is composed as follows:
{{ fullname }}-{{ replicaCount }}.{{ fullname }}-headless.{{ namespace }}.svc.cluster.local.
Since the common name of a certificate may not exceed 64 bytes, this leads to the following name length restriction:
2 * length(fullname) + length(replicaCount) + length(namespace) < 35 characters.
The sensitiveKey and clientSecret secrets cannot be passed as Kubernetes secrets. This means that the configurations should not be pushed into a version control system.
Automatic scaling of an existing NiFi cluster is not possible by simply increasing the replicaCount. Once deployed, you must additionally add the new nodes manually to the configuration of all cluster nodes.

5. Conclusion

This article gave you a tutorial on how to set up a secure Apache NiFi cluster with Apache NiFi Registry integration. It also addresses the limitations and challenges that can arise in this process. If everything worked well, you can now seamlessly dive into modeling data integration workflows and become familiar with Apache NiFi's core functionality. With this knowledge, you're well-equipped to harness the advantages of this powerful platform and effectively handle your data integration tasks.

Forem: Jannik Rebmann

Workbench for Apache NiFi data flows

Motivation

Concept

Workflow

Implementation

Process Group: Build FlowFile

Process Group: Assert FlowFile

Conclusion

Setup a secure Apache NiFi cluster in Kubernetes

Motivation

Prerequisites

1. Preparations

1.1 Enable ingress in minikube

1.2 Integrate kubectl for minikube

1.3 Install cert-manager in minikube

1.4 Map NiFi domains to minikube

1.5 Register OpenID connect (OIDC) clients

2. Setup an Apache NiFi cluster

3. Setup an Apache NiFi Registry

4. Configure Apache NiFi

4. Limitations and Troubleshooting

5. Conclusion

Process Group: `Build FlowFile`

1.2 Integrate `kubectl` for minikube