Custom Processor In Apache Nifi Using Python

Published On: 2 September 2024.By Sushil Kumar Sadhnani.

Data, AI & Analytics

Custom Processor In Apache Nifi Using Python

Introduction

Apache NiFi is a robust, open-source data integration tool that simplifies the flow of data between systems. In this blog, we will explore how to create custom processors in Apache NiFi using Python, leveraging the new capabilities introduced in NiFi 2.0. Custom processors are essential when built-in processors do not meet your specific data processing needs, and Python makes this process more accessible and flexible.

What is Apache NiFi?

Apache NiFi is an open-source platform designed to automate the flow of data between disparate systems. It provides a user-friendly web-based interface for designing, controlling, and monitoring data flows. NiFi enables seamless data movement, ensuring data is delivered from source to destination efficiently.

Purpose and Key Features of Apache NiFi

NiFi’s primary purpose is to simplify data movement, supporting both real-time and batch data processing. Key features include:

Visual Command and Control: A drag-and-drop interface for designing and managing data flows.
Data Provenance: Complete tracking of data from ingestion to output.
Backpressure: Control mechanisms to manage the flow of data and prevent system overload.
Security: Built-in security features like SSL and SSH for secure data transfer.
Extensibility: Support for custom processors, allowing users to extend NiFi’s functionality.

Core Concepts of Apache NiFi

FlowFiles: Core data packets in NiFi.
Processors: Modular units that perform operations on data.
Connections: Links between processors that define data paths and manage queues.
Flow Controller: Orchestrates the execution of processors.
Process Groups: Logical groupings of processors for better organization.

Common Use Cases of Apache NiFi

Data Ingestion and ETL: Automating data pipelines for consistent processing.
Real-time Analytics: Collecting and processing data in real time.
IoT Data Collection: Handling high-volume IoT data streams.
Log and Event Data Processing: Enhancing monitoring and operational intelligence.

Custom Processors in Apache NiFi

Custom processors in NiFi allow for tailored data processing tasks beyond the capabilities of built-in processors. With the introduction of NiFi 2.0, developers can now create custom processors using Python, which offers a more accessible development environment compared to Java.

Setting Up the Environment

Before you begin, make sure you have Python 3.9+ installed, and Apache NiFi 2.0 or later is set up. Ensure that Python support is enabled by uncommenting the nifi.python.command line in the conf/nifi.properties file.

Processor Class

To create a custom processor, you’ll need to define a Python class that extends from nifiapi.FlowFileTransform. This class will contain the logic for transforming FlowFiles.

from nifiapi.flowfiletransform import FlowFileTransform, FlowFileTransformResult

class MyCustomProcessor(FlowFileTransform):
    class Java:
        implements = ['org.apache.nifi.python.processor.FlowFileTransform']

    def __init__(self):
        super().__init__()

    def transform(self, context, flowfile):
        # Example transformation logic
        content = flowfile.getContentsAsBytes().decode('utf-8')
        modified_content = content.upper()  # Simple transformation: convert content to uppercase
        return FlowFileTransformResult(relationship="success", contents=modified_content.encode('utf-8'))

from nifiapi.flowfiletransform import FlowFileTransform, FlowFileTransformResult

class MyCustomProcessor(FlowFileTransform):

class Java:

implements = ['org.apache.nifi.python.processor.FlowFileTransform']

def __init__(self):

super().__init__()

def transform(self, context, flowfile):

# Example transformation logic

content = flowfile.getContentsAsBytes().decode('utf-8')

modified_content = content.upper() # Simple transformation: convert content to uppercase

return FlowFileTransformResult(relationship="success", contents=modified_content.encode('utf-8'))

In this example, the transform method reads the content of the FlowFile, converts it to uppercase, and writes it back to the FlowFile.

Defining Processor Details

You’ll also need to define the processor’s details, such as its version, description, tags, and dependencies. This is done using the ProcessorDetails class.

class MyCustomProcessor(FlowFileTransform):
    class ProcessorDetails:
        version = "1.0.0"
        description = "A custom processor that converts text to uppercase."
        tags = ["text", "transformation", "example"]
        dependencies = []

class MyCustomProcessor(FlowFileTransform):

class ProcessorDetails:

version = "1.0.0"

description = "A custom processor that converts text to uppercase."

tags = ["text", "transformation", "example"]

dependencies = []

These details help in identifying and managing the processor within NiFi.

Handling Lifecycle Methods

Lifecycle methods like onScheduled and onStopped are used to manage resources when the processor starts or stops.

class MyCustomProcessor(FlowFileTransform):
    def onScheduled(self, context):
        # Initialize resources, e.g., database connections, file handles
        pass

    def onStopped(self, context):
        # Cleanup resources
        pass

class MyCustomProcessor(FlowFileTransform):

def onScheduled(self, context):

# Initialize resources, e.g., database connections, file handles

pass

def onStopped(self, context):

# Cleanup resources

pass

These methods ensure that resources are properly managed throughout the processor’s lifecycle.

Adding Third-Party Dependencies

If your processor relies on third-party libraries, specify these dependencies in a requirements.txt file, or directly within the ProcessorDetails class.

Example requirements.txt:

pandas
numpy

1 2	pandas numpy

Place this file in the same directory as your processor code. NiFi will install these dependencies automatically.

Understanding Python Processor Interfaces

NiFi provides different interfaces for specific data processing needs. Here’s how you can implement these interfaces:

FlowFileTransform: For manipulating data within FlowFiles.
RecordFileTransform: For processing record-oriented data like CSV or JSON.
FlowFileSource: For creating FlowFiles from external sources.

Deploying and Documenting the Processor

Once your custom processor is developed, deploy it by copying the .py file to the $NIFI_HOME/python/extensions directory. Ensure that the processor class extends the correct NiFi interface and includes all necessary dependencies.

To document the processor, use the ProcessorDetails class to describe the processor’s purpose and usage. You can also use the @use_case decorator to provide specific examples of how the processor can be used.

from nifiapi import use_case

@use_case("Converts text to uppercase for data normalization.")
class MyCustomProcessorDetails(ProcessorDetails):
    # Processor details here

from nifiapi import use_case

@use_case("Converts text to uppercase for data normalization.")

class MyCustomProcessorDetails(ProcessorDetails):

# Processor details here

Conclusion

Apache NiFi’s support for custom Python processors opens new doors for flexible and powerful data processing workflows. By following this guide, you can create, deploy, and document your custom processors, tailoring them to your specific data transformation needs. With Python’s ease of use, extending NiFi’s capabilities has never been more straightforward.