Short Description: A NiFi connection is where FlowFiles are temporarily held between two connected NiFi Processor components. Each connection that contains queued NiFi FlowFiles will have a footprint in the JVM heap. This article will breakdown a connection to show how NiFi manages the FlowFiles that are queued in that connection and that affects heap and performance. Article: First let me share the 10,000 foot view and then I will discuss each aspect of the following image: *** NiFi FlowFiles consist of FlowFile Content and FlowFile Attributes/metadata. FlowFile content is never held in a connections heap space. Only the FlowFile Attributes/metadata is placed in heap by a connection. The "Connection Queue": The connection queue is where all FlowFiles queued in the connection are held. To understand how these queued FlowFiles affect performance and heap usage, lets start by focusing on the "Connection Queue" dissection at the bottom of the above image. The overall size of a connection is controlled by the configured "back Pressure Object Threshold" and "Back Pressure Data Size threshold" settings the user defines per connection. Back Pressure Object Threshold and Back Pressure Data Size Threshold: The "Back Pressure Object Threshold" default setting here is 10000. The "Back Pressure Data Size Threshold" defaults to 1 GB. Both of these settings are soft limits. This means that they can be exceeded. As an example, lets assume default settings above and a connection that already contains 9,500 FlowFiles. Since the connection has not reached or exceeded object threshold yet, the processor feeding that connection will be allowed to run. If that feeding processor should produce 2,000 FlowFiles when it executes, the connection would grow to 11,500 queued FlowFiles. The preceding processor would then not be allowed to execute until the queue dropped below the configured threshold once again. The same hold true for the Data Size threshold. Data Size is based on the cumulative reported size of the content associated to each queued FlowFile. Now that we know how the overall size of "connection queue" is controlled, lets break it down it to its parts: 1. ACTIVE queue: FlowFiles enter a connection will initially begin to placed in the active queue. FlowFiles will continue to placed in to this queue until that queue has reached the global configured nifi swap threshold. All FlowFiles in the active queue are held in heap memory. The processor consuming Flowfiles from this connection will always pull FlowFiles from the active queue. The size of the active queue per connection is controlled by the following property in the nifi.properties file: nifi.queue.swap.threshold=20000 Increasing the swap threshold increase the potential heap footprint of every single connection in your dataflow(s). 2. SWAP queue: Based on above default setting, once a connection reaches 20,000 FlowFiles, new FlowFiles entering the connection are placed in the swap queue. The swap queue is also held in heap and is hard coded to 10,000 FlowFiles max. If space is freed in the active queue and no swap files exist, FlowFiles in the swap queue will be moved directly to the active queue. 3. SWAP FILES: Each time the swap queue reaches 10,000 FlowFiles, a swap file is written to disk that contains those FlowFiles. At that point new FlowFiles are again written to the swap queue. Many Swap files can be created. Using image above where connection contains 80,000 FlowFiles, there would be 30,000 FlowFiles in heap and 5 swap files. As the active queue has freed 10,000 FlowFiles, the oldest swap file are moved to the active queue until all swap files are gone. The fact that swap files must be written to and read from disk, having a lot of swap files being produced across your dataflow will affect throughput performance of your dataflow(s). 4. IN-FLIGHT queue: Unlike the above 3, the in-flight queue only exists when the processor consuming from this connection is running. The consuming processor will only pull FlowFiles from the active queue and place them in the in-flight queue until processing has successfully completed and those FlowFiles have been committed to an outbound connection from the consuming processor. This in-flight queue is also held in heap. Some processors work on 1 FlowFile at a time, others work on batches of FlowFile, and some have the potential of working on every single FlowFile on an incoming connection queue. In the last case, this could mean high heap usage while those FlowFiles are being processed. The example above is one of those potential case using the MergeContent processor. The MergeContent processor places FlowFiles from the active queue in virtual bins. How many bins an what makes a bin eligible for merge is governed buy the processor configuration. What is important to understand that is is possible that every FlowFile in the connection could make its way into the "in-flight queue". In image example, if the MergeContent were running, all 80,000 queued Flowfiles would likely be pulled in to heap via the in-flight queue. ---- Take away from this article: 1. Control heap usage by limiting size of connection queue when possible. (Of course if your intention is to merge 40,000 FlowFiles, there must be 40,000 Flowfiles in the incoming connection. However, you could have two mergeContent processors in series each merging smaller bundles with same end result with less overall heap usage.) 2. With default back pressure object threshold settings, there will be no swap files produced on most connections (remember soft limits) which will result in better throughput performance. 3. The default configured swap threshold of 20,000 is a good balance in most cases of active queue size and performance. For smaller flows you may be able to push this higher and for extremely large flows you may want to set this lower. Just understand it is a trade-off of heap usage for performance. But if your run out of heap, there will be zero performance. Thank you, Matt ... View more

Short Description: This article covers how to improve the performance of the NiFi UI. Article: Over time it has been seen that the users of NiFi have been building very large dataflows consisting of many thousands of components (processor, reporting tasks, controller services, etc). While NiFi in no way limits to any degree the number of components that can be added to the NiFi canvas, the more components a user adds, the less responsive the UI becomes. This processor explosion not only affects the responsiveness of the UI, but can also lead to unexpected node disconnections. --- What are the various states a component can have? NiFi components have multiple states that consist of stopped, started, enabled, and disabled. Beyond these states exists one of two statuses: Valid: Component configuration was successfully validated. This means that all required properties have been configured and in the case of processors all required connections have been accounted for (connected to another component or terminated) and any referenced controller services have been enabled. Invalid: Component configuration is not valid. This means that one or more required properties have not been configured and/or in the case of processors one or more connections have not been accounted for (connected to another component or terminated) and/or a referenced controller services have not been enabled. --- Why does a processors state affect UI performance? All processor components when added to the canvas are added in the "stopped" state. A user can then either start or disable that component manually. All Controller Services and Reporting tasks added by a user are by default disabled. The user can then enable these components as needed. NiFi regularly must validate these components to see if they are valid or invalid. While the validation of a few hundred to a thousand components adds up to very little time, the same does not hold true for NiFi instances consisting of thousands upon thousands of components. User may have noticed a swirling on the right hand side of the NiFi status bar that seems to never go away. In a NiFi cluster, NiFi must retrieve the flow status from every node. It is possible for a component to be valid on one node but not another (for example, processor depends on local file that does not exist on all nodes). If this validation takes too long a node may be disconnected because the request took to long. Not to mention the UI does not update until these validations have completed. --- What has NiFi done to make improvements here? The bad news: Prior to NiFi 1.1.0 there is nothing that can be done to improve performance here other then reducing the number of components you are using. This is because in versions of NiFi prior to components were validated in all four states. The good news: In NiFi 1.1.0 a change was made so that this validation only occurs on components that are in the "stopped" state and controller services or reporting tasks that are disabled. It is safe to assume that if a processor is running, it must be valid. It is also safe to assume that a Controller Service or a Reporting Task must be valid if it is enabled. Now that these "started" processors and "enabled" controller services or reporting tasks are no longer being validated, the UI performance will be much better. https://issues.apache.org/jira/browse/NIFI-2996 --- What is the important to understand here? It has also been observed that users add lots of components to the UI that are never started or are only started for short periods of time. If the number of "Stopped" processors is very high, validation is still going to take a considerable amount of time even in NiFi 1.10 or newer versions. A quick look at the NiFi status bar above your canvas will show how many stopped components you have on your canvas: To make sure the UI performance remains solid, it is important that users disable processors that are not in use on the canvas. You can use the "NiFi Summary" UI to to find stopped/invalid processors and disable them. Select the "PROCESSORS" tab and sort on the "Run Status" column. Clicking on the on the right hand side of row will take you directly to that processor. Once a component is selected on the canvas it can be disabled or enabled via the "Operate" panel or by right clicking on processor and selecting Disable or Enable for displayed context menu. ... View more

The intent of this article is to show how NiFi policies in Ranger map to what you would see when using NiFi's default file based authorizer via the NiFi UI. This article will cover what access each of the policies granted to the entities (user and server) that assigned to them. There are controller level policies and component level policies in NiFi. The controller level policies are not tied to any specific component uuid. In Ranger those policies will just show as /<some policy name> - These include the following:   Ranger Policy (Base policies): NiFi Policies (Hamburger menu) Ranger permissions description: /resources *** Note: No policies will be available until this policy is manually added. N/A This policy allows Ranger to retrieve a listing of all available policies from NiFi. The server/user from the keystore being used by Ranger must be granted “read” privileges to this resource. /flow * See note [3] below View the user interface Read/View - This policy gives users the ability to view the NiFi UI. All users must be granted “read” privileges to this policy or they will not be able to open the NiFi UI. If you are running a NiFi Cluster and/or accessing Your NiFi via a proxy, You need to grant all Nodes and any proxies read access to this policy as well. Write/Modify - N/A /system View system Diagnostics Read/View - Gives granted users access to the system diagnostics. In a NiFi cluster, nodes will need to access as well to display system diagnostic stats returned by other nodes. Write/Modify - N/A /controller Access the controller Read/View - Gives granted users and/or NiFi cluster nodes the ability to view:- Controller thread pool configuration- Cluster management page- Controller level Reporting tasks- Controller level Controller services Write/Modify - Gives granted users and/or NiFi cluster nodes the ability to create/modify:- Controller thread pool configuration- Cluster management page- Controller level Reporting tasks- Controller level Controller services /counters Access counters Read/View - Gives granted users ability to view counters Write/Modify - Gives granted users ability to modify counters /provenance Query provenance Read/View -Gives granted users ability to run provenance queries or access Provenance lineage graphs. Write/Modify - N/A /restricted-components * See note [1] below Access restricted components Read/View - N/A Write/Modify - Gives granted users ability to add components to the canvas that are tagged as “restricted” /proxy * See note [2] below Proxy user requests Read/View - Allows proxy servers to send request on behalf of other users. Write/Modify - Required /site-to-site Retrieve site-to-site details Read/View - Allows Other NiFi nodes to retrieve Site-To-Site details about this NiFi. /policies *** This policy has no purpose when using ranger and does not need to be used. Access all policies Read/View - Gives granted users the ability to view existing policies. Write/Modify - Gives granted users the ability to create new policies and modify existing policies. /tenants *** This policy has no purpose when using Ranger and does not need to be used. Access users/user groups Read/View - Gives granted users the ability to view currently authorized users and user groups. Write/Modify - Gives granted uses the ability to add, delete, and modify existing users and user groups. /parameter-contexts  Access parameter contexts Read/View - Allows users to view and use ALL existing parameter contexts. Write/Modify - Allows users to create, modify, and delete ALL parameter contexts. /parameter-contexts/<uuid> Access Specific existing parameter context Read/View - Allows users to view and use a specific existing parameter context. Write/Modify - Allows users to modify or delete a specific parameter context. [1] new sub policies introduced for "/restricted-components" as of HDF 3.2 (Apache NiFi 1.12+). See following article for details: https://community.cloudera.com/t5/Community-Articles/NiFi-Restricted-Components-Policy-Descriptions/ta-p/249157 [2] All nodes in your NiFi cluster must be assigned to the "/proxy" policy. [3] All users must at a minimum be assigned to the "/flow" policy in order to view the NiFi UI. - The component level granular policies are based on the components assigned uuid. For connections, the policies are enforced based upon the processor component the connection originates from. - This includes the following policies:   Ranger Component based policies: NiFi Component based policies: component Equivalent NiFi file based authorizer policy:Policy Ranger permissions description: /data-transfer/input-ports/<uuid> Each NiFi remote input port is assigned a unique <uuid> Receive data via site-to-site Both read and write is required and should be granted to the source NIFi servers sending data to this NiFi via this input port. /data-transfer/output-ports/<uuid> Each NiFi remote output port is assigned a unique <uuid> Send data via site-to-site Both read and write is required and should be granted to the source NIFi servers pulling data from this NiFi via this output port. /process-groups/<uuid> Each NiFi process group is assigned a unique <uuid> View the component Modify the component Read - (allows user to view process group details only) Write - (allows user to start, stop or delete process group. Users are able to added components inside process group and add controller services to process group) /data/process-groups/<uuid> Each NiFi process group is assigned a unique <uuid> View the data Modify the data Read - (allows user to view data was processed by components in this process group and list queues) Write - (allows users to empty queues/purge data from queues within process group) /policies/process-groups/<uuid> *** not needed when using Ranger Each NiFi process group is assigned a unique <uuid> View the policies Modify the policies Read - N/A in Ranger Write - N/A in Ranger /processors/<uuid> Each NiFi processor is assigned a unique <uuid> View the component Modify the component Read - (Allows user to view processor configuration only) Write - (Allows user to start, stop, configure and delete processor) /data/processors/<uuid> Each NiFi processor is assigned a unique <uuid> View the data Modify the data Read - (allows user to view data processed this processor and list queues on this processors outbound connections) Write - (allows users to empty queues/purge data from this processors outbound connections) /policies/processors/<uuid> *** Not needed when using Ranger Each NiFi processor is assigned a unique <uuid> View the policies Modify the policies Read - N/A in Ranger Write - N/A in Ranger /controller-services/<uuid> Each NiFi controller services is assigned a unique <uuid> View the component   Modify the component Read - (Allows user to view controller service configuration only) Write - (Allows user to enable, disable, configure and delete controller services) /provenance-data/<component-type>/<component-UUID> Each NiFi component is assigned a unique <uuid> view provenance Read - Allows users to view provenance events generated by this component Write - N/A in Ranger /operation/<component-type>/<component-UUID> Each NiFi component is assigned a unique <uuid> operate the component Read - N/A in Ranger Write - Allows users to operate components by changing component run status (start/stop/enable/disable), remote port transmission status, or terminating processor threads There will be a unique policy for each and every component based on the specific components assigned uuid available. Component level authorizations are inherited from the parent process group when no specific processor or sub process group component level policy is set. Ranger supports the " * " wildcard when assigning policies. - In a NiFi cluster, all nodes must be granted the ability to view and modify component data in order for user to list or empty queues in processor component outbound connections. With Ranger this can be accomplished by using the a wildcard to grant all the NiFi nodes read and write to "/data/*" NiFi resource. *** Users should not be given global access to all data, but instead be restricted to specific process groups they have been granted access to. *** Also note at time of writing Ranger groups are not supported by NiFi for authorization. UPDATE: Ranger based group support was added as a new feature/capability in HDF 3.1.x ... View more

The NiFi S2S protocol is used by NiFi's Remote Process Group (RPG) components to distribute FlowFiles from one NiFi instance to another. When the target NiFi is a NiFi cluster, load-balancing of the FlowFie delivery is done across all nodes in the target NiFi cluster. The default way this works (and the only way it works in versions of NiFi previous to Apache NiFi 1.2.0 or HDF 3.0) is as follows: The RPG regularly communicates with the target NiFi cluster to get load status information about each node in the cluster. This information includes the number of currently connected nodes in the target cluster, each node's hostname, port information, and the number of total queued FlowFiles on each target NiFi node. The Source NiFi uses this information to determine a data distribution strategy for its source FlowFiles it has queued. - - - Let's assume a 4 node target NiFi cluster all reporting a zero queue count. - Each node will then be scheduled to receive 25% of the data. This means a distribution pattern of node1, node2, node3, and then, node4. - Now let's assume the same 4 node target cluster; however, node 1 and node 2 report having a queue of FlowFiles that results in the following: - Node 1 and Node 2 would get 16.67% of the data while node 3 and node 4 get 33.33% of the data. This results in a distribution pattern of node1, node2, node3, node4, node3, and then node 4. So Nodes 3 and 4 get twice the opportunity to receive data over nodes 1 and 2. Once the distribution pattern is determined, the RPG connects to the first node and starts transferring data from the incoming queue to that node for 500 milliseconds or until the queue is empty. The next run of RPG will start sending to the next node in pattern and so on. As you can see by this default distribution model, the data may not always be distributed as desired. The reason this transfer was implemented this way was for performance reasons. However, when working with very small FlowFiles, where FlowFiles come in a wide range of sizes from small to large, when a better network connection exists between one target node than another, or data comes in bursts instead of continuous flow, the load-balancing will be less than ideal. With the introduction of HDF 3.0 (Apache NiFi 1.2.0). additional configuration options were added to the RPG to control the number of FlowFiles (count), amount of data (size), and/or length of transaction time (duration) per RPG port connection. This gives users the ability to fine-tune their RPG connection to achieve better load-balancing results when dealing with lighter volume dataflows, network performance differences between nodes, etc. These new configuration options can be set as follows: Each input and output port configuration will need to be set individually. Of course, setting count to a value of 1 sounds like a good way to achieve really good load-balancing, but it will cost you in performance since only one FlowFile will be sent in each transaction. So, there will be extra overhead introduced due to the volume of new connections being opened and closed. So you may find yourself playing around with these settings to achieve your desired load-balancing to performance ratio. ---------------- How do I get better load-balancing in an older version of NiFi? The RPG will send data based on what is currently in the incoming queue per transaction. By limiting the size of that queue, you can control the max number of FlowFiles that will transfer per transaction. You can set the size of object back pressure thresholds on those incoming queues to limit the number of FlowFiles queued at any given time. This will cause FlowFiles to queue on the next upstream connection in the dataflow. If a source processor is feeding the RPG directly, try putting an updateAttribute processor between that source processor and the RPG so you have two connections. As each RPG execution runs and transfers what is on the queue, the queue will be refilled for the next transaction. Apache NiFi 1.13+ update:  In newer releases of NiFi, the ability to redistribute FlowFiles within the cluster was made much more efficient and easier through the new load-balanced connection feature. This new feature (stable in Apache NIFi 1.13+ versions) is a simple configuration change that can be done on a connection. It supports numerous strategies for redistribution of FlowFiles, but for load-balanced distribution, it offers a true round-robin capability you can't get from an RPG. ... View more