NiFi Python Developer’s Guide

Any time that a Java object must be made available to the Python API, it is made available via a proxy object. This means that in order to access a Java object from Python, we need to simply call the appropriate method on the Python proxy. When this method is called, a message is generated by the Python object and sent over the socket. That message is essentially an encoding of "Invoke method ABC on object XYZ, using arguments W, X, and Y."

For example, if we have an InputFlowFile object named flowFile and we want the filename attribute, we can do so by calling:

From the Python API perspective, this is all that is necessary. Behind the scenes, a message is written to the local socket that is an encoding of the message "Invoke the getAttribute method on the object with ID 679212, with String argument 'filename'". The Java process then receives this command, invokes the specified method on the object with the given identifier, and writes back to the socket the result of that method invocation. As a result, the Python side receives the value of the "filename" attribute.

This is important to understand, because it means that any method invocation that occurs on a Java object must be serialized, written over the socket, deserialized, and then the method can be invoked. The result must then be serialized, written over the socket, and the result deserialized on the Python side. While this is a fairly efficient process, it is not nearly as efficient as simply invoking a method natively. As a result, it is important to consider the overhead of method invocations on Java objects.

It is also important to understand that any time that an object is provided as an argument to a Python Processor, that object can only be accessed on the Python side as long as the object is made available on the Java side. Because the java side cannot store all objects indefinitely, some cleanup must happen. This cleanup happens immediately after the method is invoked.

That means that if the transform of a FlowFileTransform is called with a ProcessContext object, that object is available for use ONLY during the method invocation. As soon as the method returns (successfully or not), the object will no longer be available for use. As a result, objects provided to method invocations should not be stored for later use, such as assigning a value to self.processContext.

Referencing an object that is no longer accessible will result in an error similar to:

This indication "Target Object ID does not exist for this gateway…​" indicates that the Python code is attempting to access a Java object that is no longer accessible.

The FlowFileTransform API provides a mechanism for routing and transforming a FlowFile based on its attributes as well as its textual or binary contents. Contrast this with the RecordTransform API, which provides a mechanism for routing and transforming individual Records (such as JSON, Avro or CSV Records, for example).

In order to implement the FlowFileTransform API, a Python class must extend from the nifiapi.FlowFileTransform class and implement the transform(ProcessContext, InputFlowFile) method, which returns a FlowFileTransformResult.

Additionally, the Processor class must provide two pieces of information as subclasses: the Java interface that it implements (which will always be org.apache.nifi.python.processor.FlowFileTransform) and any details about the Processor, such as the version, a description, keywords/tags that might be associated with the Processor, etc. These will be discussed in more details below, in the ProcessorDetails and Java inner classes section.

As such, a simple implementation may look like this:

The transform method is expected to take two arguments: the context (of type nifiapi.properties.ProcessContext) and the flowfile (of type InputFlowFile).

The return type is a FlowFileTransformResult that indicates which Relationship the FlowFile should be transferred to, the updated contents of the FlowFile, and any attributes that should be added to the FlowFile (or overwritten). The relationship is a required argument. The contents is optional. If the contents of the FlowFile are not to be updated, the contents should be unspecified or should be specified as None. The original FlowFile contents should not be returned, as it will have the same effect as passing None but will be more expensive, as the contents will be written out to the FlowFile. Likewise, it is more efficient to omit the attributes unless there is any attribute to add.

While the FlowFileTransform API provides the ability to operate on a FlowFile at a time, the RecordTransform API provides developers with the opportunity to operate on a single Record at a time. For example, if a FlowFile is made up of many JSON Records, the RecordTransform Processor can be used to operate on each individual record without worrying about whether the Records are colocated or not. Implementations of this API must extend from the RecordTransform base class and must also implement the following method:

def transform(self, context, record, schema, attributemap)

returning a RecordTransformResult object.

The context object is an implementation of the same ProcessContext that is used in the FlowFileTransform Processor (see context). The record is a Python dictionary that represents the record to operate on. Regardless of whether the source of the record is JSON, CSV, Avro, or some other input format, this method is provided a Python dictionary. This makes it far simpler to operate on the data within Python and means that the code is very portable, as it can operate on any format of data.

The associated schema object is an instance of a Java object, org.apache.nifi.record.serialization.RecordSchema. This provides a schema for the data. However, calls to the schema must be made over the socket to the Java side and, as such, are expensive.

Finally, the method signature provides an attributemap. This attributemap has two methods:

getAttribute(String name) : returns the value of the FlowFile’s attribute with the given name, or None if the FlowFile does not have an attribute with that name.

getAttributes() : returns a Python dictionary whose keys are FlowFile attributes' names and whose values are the associated attribute values.

Note that these two methods are identical to those in the InputFlowFile class discussed above. This allows the attributemap to be provided to a PythonPropertyValue in order to evaluate Expression Language. For example, we might determine the name of a record’s field to use for some operation by calling:

Finally, the method must return an instance of nifiapi.recordtransform.RecordTransformResult.

The RecordTransformResult constructor takes four optional named arguments:

record : the transformed version of the Record. If the record is not supplied, or if None is supplied, the input Record will be dropped from the output.

schema : the transformed schema. If this is not specified, the schema will be inferred. However, if the schema is specified, the schema is binding, not the data. So, if a field is missing from the schema, for instance, it will be dropped from the data. And if the schema has a field in it and there’s no corresponding value in the data, the field will be assumed to have a value of None.

relationship : the name of the Relationship to route the Record to. If not specified, the value will be routed to the "success" relationship. However, the implementation may choose to expose relationships other than "success" and "failure" and route records accordingly. For example, the implementation may want to record a Record to either "valid" or "invalid."

partition : By default, all Records in a given incoming FlowFile will be written to a single output FlowFile (or, more accurately, the transformed version of the Record will be, assuming that a value of None is not returned for the result’s record field). However, we may want to partition the incoming data into separate output FlowFiles. For example, we could have incoming data that has a "country" field and want a separate output FlowFile for each country. In this case, we would return a Python dictionary for the partition argument that looks something like {'country': record['country']}. If the partition has more than one field in the dictionary, all fields in the dictionary must be the same value for two Records in order for the Records to be written to the same output FlowFile.

The FlowFileSource API provides a mechanism for creating a FlowFile and routing it based on its textual or binary contents.

In order to implement the FlowFileSource API, a Python class must extend from the nifiapi.FlowFileSource class and implement the create(ProcessContext) method, which returns a FlowFileSourceResult. Notice, that the difference between FlowFileSource’s create(ProcessContext) and FlowFileTransform’s transform(ProcessContext, InputFlowFile) methods is that the former does not expect an InputFlowFile object. That is because processors based on the FlowFileSource API are "source" processors that do not accept incoming connections but are capable of creating FlowFiles themselves.

Implementing a Processor based on FlowFileSource is very similar to implementing one based on FlowFileTransform. A simple implementation looks like this:

As mentioned above, the create method only takes one argument: the context (of type nifiapi.properties.ProcessContext).

The return type is a FlowFileSourceResult that indicates which Relationship the FlowFile should be transferred to, any attributes that should be added to the FlowFile and the contents of the FlowFile. The relationship is a required argument. Each processor based on the FlowFileSource API has a success relationship and additional relationships can be created in the Processor’s Python code. attributes and contents are both optional. If attributes is not provided, the FlowFile will still have the usual filename, path and uuid attributes, but no additional ones. If contents is not provided, a FlowFile with no contents (only attributes) will be created. In case there is no useful information to return from the create method, return None can be used instead of returning an empty FlowFileSourceResult. When create() returns with None, the processor does not produce any output. When there is nothing to return, it might be useful to yield the processor’s resources and not schedule the processor to run for the period of time defined by the processor’s Yield Duration. This can be achieved by calling context.yield_resources() from the processor’s create method right before returning None.

An important aspect of any software is the ability to configure it. With NiFi, Processors are configured by their properties. In order to expose what properties are available, a Processor must expose a PropertyDescriptor for the property. The PropertyDescriptor contains all of the information necessary in order to convey how to configure the property.

A PropertyDescriptor is created using the nifiapi.properties.PropertyDescriptor class. The constructor takes two required positional arguments: name and description. All other arguments are optional.

Typically, a Processor will have multiple Property Descriptors. These descriptors are then returned to the NiFi framework by implementing the following method in the Processor (regardless of whether it is a FlowFileTransform, a RecordTransform or a FlowFileSource):

This method returns a list of PropertyDescriptors. The typical convention is to create the Property Descriptors in the Processor’s constructor and then return them in this method, such as:

There are times, however, that Processor developer wants to allow users to specify their only properties. For example, we may allow users to enter multiple key/value pairs where the key is the name of a Record field to set and the value is the value to set it to. To accomplish this, we implement the following method:

Which returns a PropertyDescriptor. For example:

If this method is not implemented and a user adds a property other than those that are explicitly supported, the Processor will become invalid. Of course, we might also specify explicit validators that can be used, etc.

Each Processor in NiFi must route its outgoing data to some destination. In NiFi, those destinations are called "Relationships." Each Processor is responsible for declaring its Relationships.

Both the FlowFileTransform and RecordTransform Processors already have a Relationship named original and one named failure. The original relationship should not be used by implementations. This is used only by the framework and allows the input FlowFile to be passed on without modification. If the Processor cannot transform its input (because the data is not valid, for example), the Processor may route the data to the failure relationship.

By default, both implementations also have a success relationship. However, Processors may override the Relationships that it defines. It does this by implementing the following method:

This method returns a list or a set of nifiapi.relationship.Relationship objects. If this method is implemented, the success Relationship will not automatically be made available. It will need to be created and returned within this list, if it is to be used. Regardless of which Relationships are exposed by the implementation, the failure and original will always be made available.

Unlike FlowFileTransform and RecordTransform Processors, FlowFileSource Processors only have a success relationship by default. Implementations can use this relationship to route the created FlowFiles. Additional relationships can be exposed by implementing the getRelationships method. In the case of FlowFileSource implementing getRelationships does not remove the success relationship. Any relationship returned by getRelationships appears besides the success relationship.

As noted above, the ProcessorDetails and Java inner classes are important to Processors. The Java inner class must be defined on all Processors and must include a member named implements that is a list of Java interfaces that the class implements. This is important, as it allows the Py4J protocol to understand how to interact with this obect from the Java side.

The ProcessorDetails class tells NiFi about the Processor so that it can allow configuration of the Processor seamlessly through the NiFi UI. Additionally, it provides details about what is necessary in order to use the Processor. The ProcessorDetails class may have several different members:

version : The implementation version of the Processor

description : A description that can be presented in the UI to explain how the Processor is to be used. This may be more than a single sentence but should be kept as a few sentences, or a short paragraph.

tags : a list of Strings that indicates tags or keywords that are associated with the Processor. When a user adds a Processor to the NiFi canvas via the UI, users may search for keywords in order to provide discoverability. For example, if a user were to search for "CSV" any Processor whose name contains the letters "CSV" would should up. Additionally, any Processor that has a "CSV" tag would also show up.

dependencies : A list of Strings that are PyPI dependencies that the Processor depends on. The format of these strings is the same as would be provided to pip install. See Adding Third-Party Dependencies for more information.

NiFi logging works much the same way as in any other application, with one important difference. NiFi aims to make the user interface intuitive and informative, and as part of that experience will surface log messages that are appropriate. In order to accommodate this, Processors should not instantiate their own loggers. Instead, Processors should simply make use of self.logger. This will be injected into the Processor after the Processor has been created. Of course, it can’t be made available before the Processor has been created, so it cannot be accessed from within the constructor. However, it can be used anywhere else.

Often times, it is necessary to create expensive objects and reuse them instead of creating an object once, using it, and throwing it away. In order to make this simpler to handle, NiFi provides a method named onScheduled. This method is optionally implemented in the Processor. If the method is implemented, it is defined as:

Where context is a ProcessContext as described earlier. The method has no return value. This method is invoked once whenever a Processor is scheduled to run (regardless of whether it’s being started due to user input, NiFi restart, etc.).

Similarly, it is often necessary to tear down resources when they are no longer necessary. This can be accomplished by implementing the following method:

This method is called once whenever the Processor has been stopped and no longer has any active tasks. It is safe to assume that there are no longer any invocations of the transform method running when this method is called.

The context object that is the parameter of the transform(), create(), onScheduled() and onStopped() methods provides access to State Manager through the getStateManager() method. State Manager is responsible for providing a simple API for storing and retrieving state. For information on State Manager, refer to the NiFi Developer’s Guide.

The Python StateManager object returned by context.getStateManager() provides access to the underlying Java StateManager. With the Python StateManager, the state is handled in the form of Python dictionaries (as opposed to Java Maps). Just like in Java, StateManager can handle both LOCAL and CLUSTER state.

Should an error occur accessing the state using the StateManager’s methods, a StateException is thrown that can be caught and handled in the Python Processor’s code. This enables the Python developer to implement the Processor in such a way that when a state-related error occurs, the Processor can continue its operation without disruption. If a StateException is thrown by StateManager but not caught in the Processor’s code, the ProcessSession is rolled back and an error is logged.

Below is an example Processor that uses StateManager. This example assumes that the Processor is executed on primary node only, and only on one thread.

The @use_case decorator, defined in the nifiapi.documentation module can facilitate this. The decorator takes four arguments:

A single Processor may have multiple @use_case decorators.

When designing and creating Processors, it is important to keep in mind the idea of loose coupling. One Processor should not be dependent on another Processor in order to perform its task. That being said, it is often advantageous to build Processors that are designed to work well together. For example, a Processor that is able to perform a listing of files in a directory can provide an important capability in and of itself. Similarly, a Processor that is able to ingest the contents of a specific file and make that file’s contents the contents of a FlowFile is also an important capability in and of itself. But far more powerful than either of these individual capabilities is the notion of being able to compose a flow that lists all files in a directory and then ingests each of those files as a FlowFile. This is done by using a combination of the two. As such, it is important that the two Processors be able to work together in such a way that the output of the first is easily understood as the input of the second.

In this case, it makes sense to document this composition of Processors as a use case so that users can understand how to compose such a pipeline. This is accomplished by using the @multi_processor_use_case decorator. This decorator is very similar to the The @use_case Decorator but instead of a configuration element, it has a configurations element, which is a list of ProcessorConfiguration objects, where each ProcessorConfiguration object has both a processor_type, which is the name of the Processor, and a configuration that explains how to configure that particular Processor. The configuration element typically also explains how to connect outbound Relationships.

For example, we might use these decorators as such:

Note that in this case, we are able to specifically tell the user that the Filename property of FetchFile should be set to the value ${absolute.path}/${filename} because we know that the ListFile Processor will produce these attributes for us.

If one or more Processors are defined within a Python package, the package should define a requirements.txt file that declares all third-party dependencies that are necessary for any Processor in the package. The file structure will then typically look like this:

In this way, all of the requirements will be loaded from the requirements.txt file once for the package. There will be no need to load the dependencies once for ProcessorA and once for ProcessorB.

If your Processor is not a part of a Python package, its dependencies can be declared using the dependencies member of the ProcessorDetails inner class. This is a list of Strings that indicate the PyPI modules that the Processor depends on. The format is the same format expected by PyPI. This provides a convenience for declaring third-party dependencies without requiring that Processors be bundled into a package.

For example, to indicate that a Processor needs pandas installed, the implementation might look like this:

However, it is often necessary to declare a specific version of a dependency. And it may also be necessary to define multiple dependencies. We can do that in this manner:

Here, we accept any version of pandas (though the latest is preferred), and we require version 1.20.0 of numpy.

The first time that a user creates a NiFi Processor of a given type, NiFi will create a separate Python env (venv) for the Processor. It will use pip to install the specified dependencies from PyPI only into the appropriate Python environment for that Processor. Therefore, dependencies of one Processor are not made available to another Processor.

Beyond that, dependencies of one version of a Processor are not made available to other versions of the Processor. So, for example, if we have two different versions of the same Processor made available, version 0.0.1 and version 0.0.2, the dependencies that are necessary for version 0.0.1 will not be made available to version 0.0.2 unless version 0.0.2 of the Processor also declares those dependencies.

Some environments, however, cannot make use of pip for package management. In an air-gapped environment, for example, or in environments with strict security policies in place, pip may not be available. In such a case, Python processors can be packaged using the NiFi ARchive (NAR) format. This is a .zip file with the following specific layout, and uses a filename extension of .nar: