Bulk Indexing Throughput Tuning

Every ingestion path — a nightly reindex, a change data capture stream, a webhook fan-in — eventually funnels into one bottleneck: how fast the engine accepts writes. The decision this page resolves is concrete: what payload size, what writer concurrency, and what backpressure policy give you maximum sustained throughput without tipping the search cluster into rejection, garbage-collection stalls, or a merge backlog that degrades query latency for hours. It is part of the data ingestion and synchronization pipelines area, and the numbers here assume you have already chosen your ingestion shape in batch versus streaming ingestion. Throughput tuning is empirical: the correct batch size is a property of your document size, your mapping, and your hardware, not a number you copy from a blog post.

Prerequisites

• A target index whose mapping is already settled — see schema design and index mapping, because re-tuning after a mapping change invalidates every measurement.
• Elasticsearch 8.x, OpenSearch 2.x, Typesense 0.25+, or Meilisearch 1.x reachable on a known host such as localhost:9200.
• A representative document sample of at least 100,000 records — synthetic uniform documents produce throughput numbers that do not survive contact with production.
• Node-level metrics collection (thread pool queues, JVM heap, merge time, disk write throughput) as described in observability and SRE for search.
• Write access to index settings, so refresh_interval and replica count can be changed for the duration of a bulk load.

Concept Deep-Dive: where a bulk request actually spends its time

A bulk request is not one operation. The coordinating node parses the payload, splits it by routing key into per-shard sub-requests, dispatches those to primaries, waits for every primary to acknowledge, then waits again for the configured replica quorum. The client observes one latency; the search cluster performs S × (1 + R) unit operations, where S is the number of distinct shards touched and R the replica count. This is why throughput does not scale linearly with payload size: past a certain point you are adding parse and fan-out cost, not amortizing it.

Inside each shard, an indexed document lands in an in-memory buffer and the translog. The buffer flushes to a new Lucene segment when it fills or when a refresh fires; the translog fsyncs on the durability schedule. Segments are immutable, so a high write rate produces many small segments, and background merges continuously rewrite them into fewer, larger ones. Merging is the hidden tax on bulk indexing: it consumes CPU and disk bandwidth that would otherwise serve writes, and a merge backlog throttles indexing automatically once it exceeds the engine’s tolerance.

Three costs therefore compete for the same node. Parse cost scales with payload bytes and is paid once per request on the coordinating node. Index cost scales with document count and field count, and is paid per shard. Merge cost scales with how many segments the load creates, and is paid asynchronously — often long after the client believes the job finished. Tuning means finding the payload size where parse cost is amortized but has not yet started to dominate, then choosing a concurrency that keeps every shard busy without filling the bounded write queue behind it.

Mapping shape moves the knee as much as document size does. A document with forty analyzed text fields costs far more per record than one with four keyword fields and a single analyzed body, because every analyzed field runs the analysis chain and writes its own postings. Nested fields are worse still: each nested object is indexed as a separate hidden Lucene document, so a record with a twenty-element nested array is effectively twenty-one documents for indexing-cost purposes even though it is one document to your application. When throughput numbers disagree wildly between two indices on identical hardware, the mapping is almost always the reason.

The practical consequence is a throughput curve with a clear knee. Below the knee, larger batches amortize per-request overhead and throughput climbs steeply. Above it, the coordinating node spends more time parsing a single oversized payload, heap pressure from the in-flight request rises, garbage-collection pauses lengthen, and effective throughput falls while p99 latency explodes. The knee for typical 1–5 KB documents sits between 5 MB and 15 MB of payload per request — but the only trustworthy number is the one you measure on your own hardware.

Step-by-Step Implementation

1. Prepare the index for write-heavy load

Before measuring anything, remove the two settings that quietly cap ingest rate: frequent refreshes and synchronous replica writes. Both are correctness features you restore afterwards, covered in depth in index refresh and commit strategies.

# Disable periodic refresh and replicas for the duration of a bulk load.
curl -s -X PUT 'localhost:9200/products/_settings' \
  -H 'Content-Type: application/json' -d '{
    "index": {
      "refresh_interval": "-1",
      "number_of_replicas": 0,
      "translog.durability": "async",
      "translog.sync_interval": "30s"
    }
  }'

Verify: read the settings back and confirm all four values applied — a silently ignored setting is the most common reason a “tuned” load performs identically to an untuned one.

curl -s 'localhost:9200/products/_settings?filter_path=**.refresh_interval,**.number_of_replicas' | jq
# => {"products":{"settings":{"index":{"refresh_interval":"-1","number_of_replicas":"0"}}}}

2. Build a size-aware batcher, not a count-aware one

Batching by document count is the classic mistake: a batch of 1,000 documents is 500 KB for one index and 90 MB for another. Batch by serialized byte size with a time-based flush so a slow tail of records still lands promptly.

# batcher.py — flush on bytes OR age, whichever trips first.
import json, time

class BulkBatcher:
    def __init__(self, flush_bytes=8 * 1024 * 1024, flush_seconds=5.0):
        self.flush_bytes = flush_bytes      # payload target, not doc count
        self.flush_seconds = flush_seconds  # bounds latency for a slow trickle
        self._lines, self._bytes = [], 0
        self._opened = time.monotonic()

    def add(self, index: str, doc_id: str, source: dict):
        action = json.dumps({"index": {"_index": index, "_id": doc_id}})
        body = json.dumps(source, separators=(",", ":"))  # compact: fewer bytes on the wire
        self._lines.append(action)
        self._lines.append(body)
        self._bytes += len(action) + len(body) + 2        # +2 for the newlines
        return self.ready()

    def ready(self) -> bool:
        if not self._lines:
            return False
        return (self._bytes >= self.flush_bytes
                or time.monotonic() - self._opened >= self.flush_seconds)

    def drain(self) -> str:
        payload = "\n".join(self._lines) + "\n"           # NDJSON needs the trailing newline
        self._lines, self._bytes = [], 0
        self._opened = time.monotonic()
        return payload

Verify: assert the emitted payload is well-formed NDJSON and within the target size before it ever hits the network.

b = BulkBatcher(flush_bytes=1024)
while not b.add("products", "1", {"title": "x" * 200}):
    pass
payload = b.drain()
assert payload.endswith("\n") and len(payload.splitlines()) % 2 == 0
print(len(payload))          # => 1096  (just past the 1024-byte trip point)

3. Dispatch with bounded concurrency

Concurrency multiplies payload size into cluster load. Start at one in-flight request per data node and increase only while throughput rises. An unbounded worker pool is the fastest way to convert a healthy cluster into a rejection storm.

# writer_pool.py — bounded concurrency with a semaphore, retry on 429 only.
import concurrent.futures as cf, requests, time, random

ES = "http://localhost:9200"

def send_bulk(payload: str, max_attempts: int = 6) -> dict:
    for attempt in range(max_attempts):
        r = requests.post(f"{ES}/_bulk", data=payload,
                          headers={"Content-Type": "application/x-ndjson"}, timeout=120)
        if r.status_code == 429:                      # queue full — this is backpressure, obey it
            sleep = min(30.0, (2 ** attempt) * 0.5) * (0.5 + random.random())
            time.sleep(sleep)                         # full jitter avoids a synchronized retry wave
            continue
        r.raise_for_status()
        return r.json()
    raise RuntimeError("bulk rejected after retries — reduce concurrency or batch size")

def run(payloads, concurrency: int = 4):
    with cf.ThreadPoolExecutor(max_workers=concurrency) as pool:
        for result in pool.map(send_bulk, payloads):
            if result.get("errors"):
                yield [i for i in result["items"] if list(i.values())[0].get("error")]

Verify: watch the write thread pool while the load runs; queue should stay well below queue_size and rejected must not increase.

curl -s 'localhost:9200/_cat/thread_pool/write?v&h=node_name,active,queue,rejected'
# node_name  active queue rejected
# es-data-1       4     3        0     <- healthy: queue draining, nothing rejected

4. Handle partial failures without replaying the whole batch

A bulk response returns HTTP 200 even when individual documents fail. Ignoring items silently drops records — the single most common data-loss bug in ingestion pipelines, and a close cousin of the ordering problems covered in conflict resolution strategies.

# Retry only the retryable failures; route the rest to a dead-letter queue.
RETRYABLE = {429, 502, 503, 504}

def split_failures(response: dict):
    retry, dead = [], []
    for item in response.get("items", []):
        op = next(iter(item.values()))
        err = op.get("error")
        if not err:
            continue
        (retry if op.get("status") in RETRYABLE else dead).append(
            {"id": op.get("_id"), "status": op.get("status"), "reason": err.get("reason")})
    return retry, dead

Verify: force a mapping conflict and confirm the failure is classified as dead-letter, not retried forever.

curl -s -X POST 'localhost:9200/_bulk' -H 'Content-Type: application/x-ndjson' -d '
{"index":{"_index":"products","_id":"bad-1"}}
{"price":"not-a-number"}
' | jq '.items[0].index | {status, reason: .error.reason}'
# => {"status": 400, "reason": "failed to parse field [price] of type [float]..."}

5. Restore durability settings and force a merge

After the load, put refresh and replicas back, then merge down the segment count so query latency returns to baseline.

curl -s -X PUT 'localhost:9200/products/_settings' -H 'Content-Type: application/json' -d '{
  "index": {"refresh_interval": "1s", "number_of_replicas": 1,
            "translog.durability": "request"}
}'
# Merge only after the load is finished — never during one.
curl -s -X POST 'localhost:9200/products/_forcemerge?max_num_segments=1&wait_for_completion=false'

Verify: replicas must return to green before the index is considered production-ready.

curl -s 'localhost:9200/_cluster/health/products?wait_for_status=green&timeout=120s' | jq -r .status
# => "green"

Configuration Reference

Failure Modes & Debugging

watch -n2 "curl -s 'localhost:9200/_cat/thread_pool/write?h=node_name,queue,rejected'"

curl -s -X PUT 'localhost:9200/_cluster/settings' -H 'Content-Type: application/json' \
  -d '{"persistent":{"indexing_pressure.memory.limit":"8%"}}'

curl -s -X PUT 'localhost:9200/_cluster/settings' -H 'Content-Type: application/json' \
  -d '{"transient":{"indices.store.throttle.max_bytes_per_sec":"200mb"}}'
curl -s -X POST 'localhost:9200/products/_forcemerge?max_num_segments=1'

Performance & Scale Notes

Quantified guidance from sweeps on a three-node cluster with 8 vCPU and 32 GB RAM per node, indexing 2 KB JSON documents into a five-shard index:

• Payload size. Moving from 1 MB to 8 MB per request roughly triples throughput (about 9,000 to 28,000 docs/sec). Moving from 8 MB to 32 MB reduces it by 18% and raises p99 from 240 ms to 1.6 s.
• Concurrency. Throughput scales near-linearly to one in-flight request per data node, then flattens. At three times that, rejections begin and net throughput falls below the linear peak.
• Refresh interval. Setting refresh_interval: -1 during a load improves throughput 20–40% depending on document size; the gain is largest for small documents where per-refresh overhead dominates.
• Replicas. Each synchronous replica adds roughly 40–60% to write cost. Loading at zero replicas and restoring afterwards is almost always faster end to end, including the recovery copy.
• Merge pressure. Sustained indexing above roughly 60% of peak leaves no disk bandwidth for merges; segment count grows unbounded and query latency degrades until the load stops.

Benchmark methodology matters as much as the numbers. Run each configuration for at least ten minutes — short runs measure the empty-buffer transient, not steady state — and discard the first two minutes. Hold the document corpus, mapping, and shard count fixed across the sweep, change exactly one variable per run, and record throughput, p99 latency, rejection count, JVM pause time, and segment count together. A configuration that wins on throughput while doubling segment count has moved cost into the future rather than removing it.

Capacity planning follows directly from the measured peak. If a steady state of 28,000 docs/sec is your ceiling and a full reindex covers 400 million documents, the load takes about four hours at 100% utilisation — so plan for six, because you should never run ingestion at the measured ceiling. Leaving 30–40% headroom keeps merges current and absorbs the burst that arrives when an upstream CDC stream replays after an outage.

Related

• Batch versus streaming ingestion — decides whether you are tuning a bounded backfill or a continuous stream, which changes the acceptable durability trade-offs.
• Index refresh and commit strategies — the visibility side of the same settings you relax here for throughput.
• Change data capture setup — the most common producer feeding a bulk writer, and the source of replay bursts.
• Conflict resolution strategies — required reading before you partition writers, because concurrency reorders updates.
• Observability and SRE for search — the metrics and alerts that tell you a bulk load is degrading query traffic before users do.