Bulk (gRPC)

Introduced 3.0

This is an experimental feature and is not recommended for use in a production environment. For updates on the progress of the feature or if you want to leave feedback, see the associated GitHub issue.

The gRPC Bulk API provides an efficient, binary-encoded alternative to the HTTP Bulk API for performing multiple document operations—such as indexing, updating, and deleting—in a single call. This service uses protocol buffers and mirrors the REST API in terms of parameters and structure.

Prerequisite

To submit gRPC requests, you must have a set of protobufs on the client side. For ways to obtain the protobufs, see Using gRPC APIs.

gRPC service and method

gRPC Document APIs reside in the DocumentService.

You can submit bulk requests by invoking the Bulk gRPC method within the DocumentService. The method takes in a BulkRequest and returns a BulkResponse.

Document format

In gRPC, documents must be provided and returned as bytes. Use Base64 encoding to provide documents in a gRPC request.

For example, consider the following document in a regular Bulk API request:

"doc":  "{\"title\": \"Inception\", \"year\": 2010}"

For a gRPC Bulk API request, provide the same document in Base64 encoding:

"doc": "eyJ0aXRsZSI6ICJJbmNlcHRpb24iLCAieWVhciI6IDIwMTB9"

BulkRequest fields

The BulkRequest message is the top-level container for a gRPC bulk operation. It accepts the following fields.

BulkRequestBody fields

The BulkRequestBody message represents a single document-level operation within a BulkRequest. It accepts the following fields. All fields are optional, but exactly one of index, create, update, or delete must be set in the BulkRequestBody.

Create

CreateOperation adds a new document only if it doesn’t already exist.

The document itself must be provided in the object field, outside of the CreateOperation message.

The following optional fields can also be provided.

Example request

The following example shows a bulk request with a create operation. It creates a document with the ID tt1375666 in the movies index. The document content, provided in Base64 encoding, represents {"title": "Inception", "year": 2010}:

{
  "index": "movies",
  "request_body": [
    {
      "create": {
        "index": "movies",
        "id": "tt1375666"
      },
      "object": "eyJ0aXRsZSI6ICJJbmNlcHRpb24iLCAieWVhciI6IDIwMTB9"
    }
  ]
}

Delete

The DeleteOperation removes a document by ID. It accepts the following fields.

Example request

The following example shows a bulk request with a delete operation. It deletes a document with the ID tt1392214 from the movies index:

{
  "index": "movies",
  "request_body": [
    {
      "delete": {
        "index": "movies",
        "id": "tt1392214"
      }
    }
  ]
}

Index

The IndexOperation creates or overwrites a document. If an ID is not provided, one is generated.

The document itself is provided in the doc field, outside of the IndexOperation message.

The following optional fields can also be provided.

Example request

The following example shows a bulk request with an index operation. It indexes a Base64-encoded document with the ID tt0468569 into the movies index:

{
  "index": "movies",
  "request_body": [
    {
      "index": {
        "index": "movies",
        "id": "tt0468569"
      },
      "doc": "eyJ0aXRsZSI6ICJUaGUgRGFyayBLbmlnaHQiLCAieWVhciI6IDIwMDh9"
    }
  ]
}

Update

The UpdateOperation performs partial document updates.

The document itself is provided in the doc field, outside of the UpdateOperation message.

All UpdateOperation fields, listed in the following table, are optional except for id.

Example request

The following example shows a bulk request with an update operation. It will update a document with the ID tt1375666 in the movies index to {"year": 2011}:

{
  "index": "movies",
  "request_body": [
    {
      "update": {
        "index": "movies",
        "id": "tt1375666"
      },
      "doc": "eyJ5ZWFyIjogMjAxMX0=",
      "detect_noop": true
    }
  ]
}

Upsert

The upsert operation updates the document if it already exists. Otherwise, it creates a new document using the provided document content.

To upsert a document, provide an UpdateOperation but specify doc_as_upsert as true. The document to be upserted should be provided in the doc_as_upsert field outside of the UpdateOperation.

Example request

The following example shows a bulk request with an upsert operation. It updates the year field of the document with ID tt1375666 in the movies index to {"year": 2012}:

{
  "index": "movies",
  "request_body": [
    {
      "update": {
        "index": "movies",
        "id": "tt1375666"
      },
      "doc": "eyJ5ZWFyIjogMjAxMn0=",
      "doc_as_upsert": true
    }
  ]
}

Script

Run a stored or inline script to modify a document.

To specify a script, provide an UpdateOperation and a script field outside of the UpdateOperation.

Example request

The following example shows a bulk request with a script operation. It increments the year field of the document with the ID tt1375666 in the movies index by 1:

{
  "index": "movies",
  "request_body": [
    {
      "update": {
        "index": "movies",
        "id": "tt1375666"
      },
      "script": {
        "source": "ctx._source.year += 1",
        "lang": "painless"
      }
    }
  ]
}

Response fields

The gRPC Bulk API provides the following response fields.

BulkResponseBody fields

The BulkResponse message wraps either a BulkResponseBody for successful requests, or a BulkErrorResponse for failed requests. The BulkResponseBody provides a summary and per-item result of a bulk operation and contains the following fields.

Item fields

Each Item in the response corresponds to a single operation in the request. For each operation, only one of the following fields is provided.

ResponseItem fields

Each ResponseItem corresponds to a single operation in the request. It contains the following fields.

Example response

{
  "bulkResponseBody": {
    "errors": false,
    "items": [
      {
        "index": {
          "id": {
            "string": "2"
          },
          "index": "my_index",
          "status": 201,
          "primaryTerm": "1",
          "result": "created",
          "seqNo": "0",
          "shards": {
            "successful": 1,
            "total": 2
          },
          "version": "1",
          "forcedRefresh": true
        }
      },
      {
        "create": {
          "id": {
            "string": "1"
          },
          "index": "my_index",
          "status": 201,
          "primaryTerm": "1",
          "result": "created",
          "seqNo": "0",
          "shards": {
            "successful": 1,
            "total": 2
          },
          "version": "1",
          "forcedRefresh": true
        }
      },
      {
        "update": {
          "id": {
            "string": "2"
          },
          "index": "my_index",
          "status": 200,
          "primaryTerm": "1",
          "result": "updated",
          "seqNo": "1",
          "shards": {
            "successful": 1,
            "total": 2
          },
          "version": "2",
          "forcedRefresh": true,
          "get": {
            "found": true,
            "seqNo": "1",
            "primaryTerm": "1",
            "source": "e30="
          }
        }
      },
      {
        "delete": {
          "id": {
            "string": "2"
          },
          "index": "my_index",
          "status": 200,
          "primaryTerm": "1",
          "result": "deleted",
          "seqNo": "2",
          "shards": {
            "successful": 1,
            "total": 2
          },
          "version": "3",
          "forcedRefresh": true
        }
      }
    ],
    "took": "87",
    "ingestTook": "0"
  }
}

Java gRPC client example

The following example shows a Java client-side program that submits a sample bulk gRPC request and then checks whether there were any errors in the bulk response:

import org.opensearch.protobufs.*;
import io.grpc.ManagedChannel;
import io.grpc.ManagedChannelBuilder;
import com.google.protobuf.ByteString;

public class BulkClient {
    public static void main(String[] args) {
        ManagedChannel channel = ManagedChannelBuilder.forAddress("localhost", 9400)
                .usePlaintext()
                .build();

        DocumentServiceGrpc.DocumentServiceBlockingStub stub = DocumentServiceGrpc.newBlockingStub(channel);

        IndexOperation indexOp = IndexOperation.newBuilder()
                .setIndex("my-index")
                .setId("1")
                .build();

        BulkRequestBody indexBody = BulkRequestBody.newBuilder()
                .setIndex(indexOp)
                .setDoc(ByteString.copyFromUtf8("{\"field\": \"value\"}"))
                .build();

        DeleteOperation deleteOp = DeleteOperation.newBuilder()
                .setIndex("my-index")
                .setId("2")
                .build();

        BulkRequestBody deleteBody = BulkRequestBody.newBuilder()
                .setDelete(deleteOp)
                .build();

        BulkRequest request = BulkRequest.newBuilder()
                .setIndex("my-index")
                .addRequestBody(indexBody)
                .addRequestBody(deleteBody)
                .build();

        BulkResponse response = stub.bulk(request);
        System.out.println("Bulk errors: " + response.getErrors());

        channel.shutdown();
    }
}