Tunable Parameters in the setup of NiFi and Elasticsearch

System resources, memory footprint, and CPU allocation

The environment (recommended or minimal) will constitute of several nodes that will host the NiFi and Elasticsearch pods. In general, you will have after the initial installation, three pods for Elasticsearch and one NiFi.

In the minimal configuration, each of the pods is allocated six vCPUs and 10GB of memory.

In the recommended configuration, each of the pods is allocated sixteen vCPUs and a minimum of sixteen GB of memory.

Adjustment or increase of the allocated resources is possible. In these cases, however, additional testing to validate such configuration changes is nessessary, to ensure stability and operability.

Disk space per node

The amount of disk space you require is strongly dependent on the size of your index. Since this varies widely from installation to installation, there is no single recommended figure. To estimate, consider that an index of one million catalog items running on the recommended system configuration generates an index about six gigabytes in size.

This number does not reflect your actual requirements, however. Index files are dated and with time will accumulate on the disk. As a general rule, provide at least ten times the disk space of the anticipated index size to reflect this variability, and clean up old index files on a daily basis. In the case of the six gigabyte index, this would mean allocating at least sixty gigabytes and running a regular job to delete stale files.

NiFi

The processing speed of the data set, and the resulting speed of the search index creation is the result of the throughput one can achieve in the NiFi and Elasticsearch cluster. There are several parameters that would improve and optimize the throughput for a given hardware footprint; NiFi processors threads and bucket size.

Threads

The default processor runs on a single thread, processing just one flow file at a time. If you use concurrent processing, the number of concurrent tasks that it can do can be adjusted. Set the number of threads for the process group by changing the processor Concurrent Tasks value (under Processor configuration or SCHEDULING tab).

When the processor can process flow files at the same rate as they come, the Concurrent Tasks value is ideal, preventing large pileups of flow files in the processor's wait queue.

If a CPU can multitask, increasing the threads available to the processor increases the processor throughput. The transformation processor (as in NLP) and the Bulk update processor are two such examples. You can assign more threads to the processor, or set the Concurrent Tasks variable to be higher than one. Increasing the threads of the processor will result in improvement of the processor throughput. The following screenshot represents an NLP Processor which is set to sixteen concurrent tasks, equal to the number of virtual CPUs (vCPUs) that are available on the node.

This update is not useful for all processors. Most processors come with an default configuration that takes this variable into account and does not need to be altered. When performance testing reveals a bottleneck in front of the processors, the default configuration may benefit from further tuning. Because such balancing may not always be feasible, the recommended best practice is to focus on reducing the flow file pileup in the waiting queue.

Additional threads increase the processor bandwidth by the factor of the number of threads, if the processor is doing computational processing (that is, you can experience linear scaling). In the case of the I/O operations, the processor will experience some improvement that would start to diminish after certain level of threads is set (such as nonlinear scalability that would end in saturation).

In the case of the NLP Processor, the processor is purely computational, and the thread limit is considered to be the same number of vCPUs that are available for the NiFi Pod.

Bucket size

Bucket size (scroll.bucket.size) is another parameter you can change to improve a processor bandwidth. It changes the size of the flowfile that is processed. By increasing the bucket size you are increasing the size of data that will be processed by a single processor as a group.

Bucket size changes are a bit more difficult to implement. The location of the variable is on the second level parameters of the processor group.

For example, for DataBaseProductStage1a:

The top level appears like this:

Drill down one level, right-click on Create Product Document and select variables.

The Bucket Size is optimal when you see that the flow file is easily processed through the system (including Elasticsearch upload). Increasing the size (number of documents in the flow file) increases the throughput. However, when increased to very high size, the throughput tapers off and gradually decreases. Simultaneously, the resources requirements of the system (NiFi and Elasticsearch) increase, leading to lower throughput.

The Variables window opens. Here you can change the bucket size value.

Scroll page size (scroll.page.size)

scroll.page.size

The scroll.page.size parameter contains the number of rows fetched from database by the SQL. SQL runs several times (once with each page turn) if the retrieved row count is smaller than the total row count in the table.

When the time it takes to process the data is evenly matched with the time it takes for the next SQL scroll to receive its data, the scroll settings are optimal. This improvement eliminates needless processing and I/O time. Additional factors to consider include the memory space available in NiFi to hold the result set while parsing and splitting this into flowfiles, and the total number of flow files that would be created in NiFi at once. If the scroll page size is larger, you should expect an impact on NiFi operations. When you reach this limit, you can increase the resources allocated to NiFi, or limit the scroll page size to reduce the impact to performance.

LISTAGG() and Serialize

In the ingest pipeline, most SQLs are using the aggregate to combine multiple rows together. Due to a limitation of the database (especially with Oracle), and the size of the data, LISTAGG can exceed the limits in place during the live.reindex and auth.reindex index builds. To resolve the issue, you can disable the LISTAGG.

You can set LISTAGG locally or globally. To set it locally, change the attribute flow.database.listagg.

You control attributes using UpdateAttribute processors, which update the attributes of the flow files. For example, if you want to set flow.database.listagg="false" for AttributeStage1b in auth.reindex, set it in the Properties as follows: NiFi Flow > auth.reindex - Attribute Stage 1b (Find Attribute Values) > Find Attribute Values > SCROLL SQL > Define scroll offset and page size.

Note: If you experience issues during a particular stage of ingest pipeline processing where string aggregation is exceeding the 32k LISTAGG function limit, you will need to disable LISTAGG for that particular processing stage. For instance, to disable LISTAGG for Attribute Stage 1b (Find Attribute Values) in versions prior to 9.1.11:

1. Go to NiFi Flow > auth.reindex - Attribute Stage 1b (Find Attribute Values) > Find Attribute Values > SCROLL SQL > Define scroll offset and page size.
2. Double-click on the Define scroll offset and page size processor.
3. Click STOP & CONFIGURE in the upper right-hand corner of the processor to stop the processor.
4. Click the Properties tab of the processor and then click the + icon to add the flow.database.listagg property and set its value to false.
5. Restart the processor.

For versions 9.1.11 and up, use the following endpoint:

https://data-query/search/resources/api/v2/configuration?nodeName=ingest&envType=auth

Where data-query is the URL for the Query server. Use the GET method to retrieve Ingest configuration information. Use the PATCH method to set it false, with the following body to the query:

{
    "global": {
        "connector": [
            {
                "name": "attribute",
                "property": [                   
                    {
                        "name": "flow.database.listagg",
                        "value": "false"
                    }
                ]
            }
        ]
    }
}

You can make make global changes using Ingest profiles. For more information and an example of how to change LISTAGG using an Ingest profile, see Ingest configuration via REST.

When the list aggregate is disabled, the SQL proccess will return more rows and NiFi will use the Serialize process to handle the returned data. In the case, the duration for processing the data will be much longer. To account for this, the page.size and bucket.size for the SQL process, and thread number for the Serialize process must be increased.

Elasticsearch

The following section will discuss a few improvements that can be made to the Elasticsearch configuration to increase the overall throughput and speed of index building.

Sharding