This processor is for accessing the Elasticsearch Bulk API. It provides the ability to configure bulk operations on a per-record basis which is what separates it from PutElasticsearchJson. For example, it is possible to define multiple commands to index documents, followed by deletes, creates and update operations against the same index or other indices as desired.

As part of the Elasticsearch REST API bundle, it uses a controller service to manage connection information and that controller service is built on top of the official Elasticsearch client APIs. That provides features such as automatic master detection against the cluster which is missing in the other bundles.

This processor builds one Elasticsearch Bulk API body per record set. Care should be taken to split up record sets into appropriately-sized chunks so that NiFi does not run out of memory and the requests sent to Elasticsearch are not too large for it to handle. When failures do occur, this processor is capable of attempting to write the records that failed to an output record writer so that only failed records can be processed downstream or replayed.

The index, operation and (optional) type fields are configured with default values that can be overridden using record path operations that find an index or type value in the record set. The ID and operation type (create, index, update, upsert or delete) can also be extracted in a similar fashion from the record set. An "@timestamp" field can be added to the data either using a default or by extracting it from the record set. This is useful if the documents are being indexed into an Elasticsearch Data Stream. The following is an example of a document exercising all of these features:

        {
            "metadata": {
                "id": "12345",
                "index": "test",
                "type": "message",
                "operation": "index"
            },
            "message": "Hello, world",
            "from": "john.smith",
            "ts": "2021-12-03'T'14:00:00.000Z"
        }

        {
            "metadata": {
                "id": "12345",
                "index": "test",
                "type": "message",
                "operation": "delete"
            }
        }

The record path operations below would extract the relevant data:

- /metadata/id
- /metadata/index
- metadata/type
- metadata/operation
- /ts

Valid values for "operation" are:

- create
- delete
- index
- update
- upsert