Bulk Document Indexing and Batch Processing in OpenSearch: A Complete Tutorial

Bulk document indexing in OpenSearch involves accumulating documents in memory and flushing them in sized batches—typically 500 documents (1,000 bulk actions) for standard indexing and 20 items for vectorized workflows—to optimize throughput while managing API latency and memory usage.

The aws-samples/sample-for-amazon-opensearch-service-tutorials-101 repository demonstrates production-ready patterns for bulk document indexing and batch processing in OpenSearch. The implementation provides two complementary approaches: standard bulk ingestion for regular JSON documents and vectorized batch processing that generates embeddings via Amazon Bedrock before indexing.

Standard Bulk Indexing Approach

The standard bulk indexing workflow handles high-throughput ingestion of plain JSON documents into a traditional OpenSearch index named products.

Index Creation and Setup

Before writing data, the system ensures the target index exists with appropriate mappings. The create_index() function in artifacts/index_lambda/opensearch_index.py defines text fields with a stop analyzer and numeric fields as float types.

create_index()  # ensures the "products" index exists with proper mappings

This automatic index creation prevents write errors and establishes the correct field types for search and aggregation operations.

Batch Size and Chunking Logic

The bulk_index_documents() function implements memory-efficient batch processing by accumulating documents in Python lists and flushing them when reaching specific thresholds.

bulk_data.append({"index": {"_index": INDEX_NAME, "_id": f"{uuid.uuid4().hex}"}})
bulk_data.append(doc)

The implementation flushes the buffer every 500 documents (1,000 bulk actions, since each document requires both an action line and a source line). This chunk size balances network efficiency with Lambda memory constraints. After the loop completes, any remaining documents are flushed to ensure no data loss.

Error Handling and Response Validation

The function inspects the bulk response for errors after each flush. If the response contains error details, the function returns a failure payload with diagnostic information, allowing upstream API handlers to return appropriate HTTP status codes.

Vectorized Bulk Indexing with Embeddings

For semantic search use cases, the repository implements vectorize_and_index_products(), which generates 1024-dimensional embeddings via Amazon Bedrock before writing to specialized k-NN indices.

Search Pipeline and Index Configuration

The workflow first creates a search pipeline using search_nlp() and establishes two separate vector indices with different storage modes:

• products_vectorized_on_disk – persistent storage for cost-effective large-scale datasets
• products_vectorized_in_memory – high-performance storage for low-latency queries

The indices are created via create_vector_index_on_disk_mode() and create_vector_index_in_memory_mode() respectively.

Bedrock Batch Processing Strategy

Unlike the standard bulk approach, vectorized processing uses a smaller batch size of 20 items to manage Bedrock API latency and avoid throttling. The implementation reads the products_content.jsonl file and processes records in chunks:

batch_size = 20
for i in range(0, len(product_list), batch_size):
    batch = product_list[i:i + batch_size]

For each batch, the code concatenates relevant text fields, invokes get_embedding() to retrieve vectors from Amazon Bedrock, and attaches the embedding to the product document.

Dual Index Bulk Submission

The same enriched document is queued for both vector indices simultaneously. The implementation builds two separate bulk payloads:

bulk_data_on_disk.append({"index": {"_index": VECTOR_INDEX_NAME_ON_DISK, "_id": f"{uuid.uuid4().hex}"}})
bulk_data_on_disk.append(product)
bulk_data_in_memory.append({"index": {"_index": VECTOR_INDEX_NAME_IN_MEMORY, "_id": f"{uuid.uuid4().hex}"}})
bulk_data_in_memory.append(product)

Each payload is sent using ops_client.bulk(), allowing direct performance comparison between on-disk and in-memory k-NN storage modes.

API Endpoints and Implementation

The Lambda handler in artifacts/index_lambda/opensearch_index.py exposes these workflows through HTTP endpoints:

• POST /index → index_products() → bulk_index_documents()
• POST /vectorize-index → vectorize_and_index_products()

Standard Bulk Indexing via API

curl -X POST https://<api-id>.execute-api.<region>.amazonaws.com/prod/POST/index \
  -H "Content-Type: application/json" \
  -d '[
        {"title":"Red Running Shoes","category":"men","price":79.99,"description":"Lightweight..."},
        {"title":"Blue Backpack","category":"accessories","price":49.99,"description":"Water‑resistant..."}
      ]'

The request body accepts an array of product objects. The Lambda wrapper converts this to a bulk request and returns a JSON payload indicating success or failure.

Vectorized Bulk Indexing via API

curl -X POST https://<api-id>.execute-api.<region>.amazonaws.com/prod/POST/vectorize-index

This endpoint reads the pre-populated products_content.jsonl catalog, generates embeddings via Bedrock, and bulk-indexes the results into both on-disk and in-memory vector indices.

Direct Python Implementation

You can invoke these functions directly within your own Python applications:

from artifacts.index_lambda.opensearch_index import bulk_index_documents, vectorize_and_index_products

# Example: bulk index a small list

docs = [
    {"title": "Green Socks", "category": "men", "price": 12.99, "description": "Cotton"},
    {"title": "Yellow Hat", "category": "women", "price": 22.49, "description": "Straw"},
]
response = bulk_index_documents(docs)
print(response)   # => {"success": True, "result": "Products indexed successfully", ...}

# Example: vectorize & index the whole product catalog

response = vectorize_and_index_products({})
print(response)   # => {"success": True, "result": "Products vectorized and indexed successfully", ...}

Summary

• Standard bulk indexing accumulates documents in memory and flushes them in chunks of 500 documents (1,000 bulk actions) to balance throughput and Lambda memory constraints.
• Vectorized batch processing uses smaller batches of 20 items to manage Bedrock embedding API latency, generating 1024-dimensional vectors before writing to k-NN indices.
• Dual index strategy simultaneously populates both on-disk (products_vectorized_on_disk) and in-memory (products_vectorized_in_memory) vector indices for performance comparison.
• Implementation location resides in artifacts/index_lambda/opensearch_index.py, with API endpoints exposed through API Gateway and Lambda handlers.

Frequently Asked Questions

What is the optimal batch size for bulk indexing in OpenSearch?

The sample application uses 500 documents per batch (1,000 bulk actions) for standard indexing, which provides an optimal balance between network efficiency and AWS Lambda memory constraints. For vectorized workflows involving Amazon Bedrock embeddings, the batch size is reduced to 20 items to prevent API throttling and manage embedding generation latency.

How does the application handle errors during bulk indexing?

The bulk_index_documents() function inspects the response from ops_client.bulk() after each flush operation. If the response contains error details, the function immediately returns a failure payload with diagnostic information, allowing the API Gateway endpoint to return an appropriate HTTP error status rather than silently dropping failed documents.

What is the difference between the on-disk and in-memory vector indices?

The tutorial creates two separate k-NN indices to demonstrate different storage modes. products_vectorized_on_disk persists vectors to disk for cost-effective storage of large datasets, while products_vectorized_in_memory keeps vectors in memory for low-latency semantic search queries. Both indices receive identical documents with 1024-dimensional embeddings generated via Amazon Bedrock.

Can I index documents without generating vector embeddings?

Yes. The POST /index endpoint invokes bulk_index_documents(), which writes plain JSON documents to the standard products index without any embedding generation. This approach is suitable for traditional keyword search and filtering use cases where semantic vector search is not required.

{
  "mcpServers": {
    "instagit": {
      "command": "npx",
      "args": ["-y", "instagit@latest"]
    }
  }
}