Tuning Bulk Request Size and Concurrency

Your ingestion job runs at 6,000 documents per second and you have no idea whether that is the ceiling or a tenth of it. Guessing at batch_size = 1000 and workers = 8 is how most pipelines are configured, and it is why most pipelines leave half their throughput on the table or quietly saturate a node. This page runs a reproducible two-dimensional sweep — payload bytes × writer concurrency — and reads a defensible operating point off the results. It belongs to bulk indexing throughput tuning inside the data ingestion and synchronization pipelines area.

Prerequisites

1. A disposable index on the target hardware with the production mapping applied — not a simplified test mapping.
2. At least 500,000 representative documents available locally as newline-delimited JSON.
3. Shell access to read _cat/thread_pool and _nodes/stats during a run.
4. A quiet cluster: no concurrent query load, no other ingestion, no ongoing shard relocation.
5. Roughly 90 minutes — a credible sweep is 12 to 16 runs of five minutes each.

Why a sweep beats a rule of thumb

Published advice (“5–15 MB per bulk request”) describes a distribution, not your index. Two variables move the optimum by more than an order of magnitude: average serialized document size and the number of analyzed fields in your mapping. A 400-byte log line with three keyword fields and a 12 KB product document with nested variants and eight analyzed text fields have almost nothing in common from the indexing engine’s point of view, even though both are “one document” to your application.

The second reason is that payload size and concurrency are not independent. In-flight bytes are roughly payload × concurrency × (1 + replicas), and it is the product that pressures heap. A configuration of 4 MB × 16 writers and one of 32 MB × 2 writers push similar byte volume but behave completely differently: the first spreads parse cost across many small requests and keeps every shard busy, while the second creates heavy-tailed latencies and heap spikes. Sweeping one variable while holding the other at a bad value produces a confident, wrong answer.

Solution Steps

1. Freeze the variables you are not measuring

Every run must differ in exactly one dimension. Fix refresh, replicas, and durability up front so they cannot drift between runs, and recreate the index between runs so segment state never carries over.

# reset_index.sh — run before EVERY sweep iteration
curl -s -X DELETE 'localhost:9200/bench' > /dev/null
curl -s -X PUT 'localhost:9200/bench' -H 'Content-Type: application/json' -d '{
  "settings": {
    "number_of_shards": 5,          
    "number_of_replicas": 0,        
    "refresh_interval": "-1",       
    "translog.durability": "async"  
  },
  "mappings": { "_source": {"enabled": true} }
}' > /dev/null

2. Drive the sweep from a harness that records everything

Throughput alone is not enough to pick a winner. Record rejections, p99 latency, and JVM pause time in the same row, because the fastest configuration that also rejects requests is not a valid operating point.

# sweep.py — two-dimensional sweep with per-run metrics
import itertools, json, statistics, subprocess, time
import concurrent.futures as cf, requests

ES = "http://localhost:9200"
PAYLOADS = [2, 8, 16, 32]        # MB per bulk request
CONCURRENCY = [2, 4, 8, 16]      # in-flight requests
RUN_SECONDS = 300                # 5 minutes; discard the first 120 s as warm-up

def node_counter(path):
    stats = requests.get(f"{ES}/_nodes/stats/thread_pool,jvm", timeout=10).json()
    return sum(_dig(n, path) for n in stats["nodes"].values())

def _dig(node, path):
    cur = node
    for k in path.split("."):
        cur = cur[k]
    return cur

def run_case(mb, workers, corpus):
    subprocess.run(["./reset_index.sh"], check=True)
    rej0 = node_counter("thread_pool.write.rejected")
    gc0 = node_counter("jvm.gc.collectors.young.collection_time_in_millis")
    lat, docs, t0 = [], 0, time.monotonic()
    with cf.ThreadPoolExecutor(max_workers=workers) as pool:
        futures = set()
        for payload, count in corpus.batches(mb * 1024 * 1024):
            if time.monotonic() - t0 > RUN_SECONDS:
                break
            futures.add(pool.submit(_post, payload))
            docs += count
            if len(futures) >= workers * 2:                # keep the pipe full, not flooded
                done, futures = cf.wait(futures, return_when=cf.FIRST_COMPLETED)
                lat.extend(f.result() for f in done)
    elapsed = time.monotonic() - t0
    return {
        "payload_mb": mb, "workers": workers,
        "docs_per_sec": round(docs / elapsed),
        "p99_ms": round(statistics.quantiles(lat, n=100)[98]) if lat else None,
        "rejected": node_counter("thread_pool.write.rejected") - rej0,
        "gc_ms": node_counter("jvm.gc.collectors.young.collection_time_in_millis") - gc0,
    }

def _post(payload: str) -> float:
    t = time.monotonic()
    r = requests.post(f"{ES}/_bulk", data=payload,
                      headers={"Content-Type": "application/x-ndjson"}, timeout=300)
    r.raise_for_status()
    return (time.monotonic() - t) * 1000

3. Read the winner off the results, not off the peak

Sort by throughput, then reject any row with a non-zero rejection count or a p99 beyond your ingestion latency budget. The operating point is the best surviving row backed off by one step, which leaves headroom for production’s noisier conditions.

rows = [run_case(mb, w, corpus) for mb, w in itertools.product(PAYLOADS, CONCURRENCY)]
valid = [r for r in rows if r["rejected"] == 0 and r["p99_ms"] < 2000]
best = max(valid, key=lambda r: r["docs_per_sec"])
print(json.dumps(best, indent=2))
# {
#   "payload_mb": 8, "workers": 8,
#   "docs_per_sec": 28400, "p99_ms": 612, "rejected": 0, "gc_ms": 3180
# }

4. Back off and pin the operating point

Run production one step below the measured peak. The sweep ran on a quiet cluster; production has query traffic, merges, and the occasional replica recovery competing for the same threads.

# ingest-config.yaml — derived from the sweep, with provenance in the file
bulk:
  payload_bytes: 8388608      # 8 MB — sweep peak
  concurrency: 4              # one step below the peak of 8, for production headroom
  max_retries: 6
  measured_on: "2026-07-26 / 3 nodes / 8 vCPU / 2 KB docs / 5 shards"

5. Re-measure when anything structural changes

A sweep result has a shelf life. It is valid for the mapping, shard count, hardware, and document mix it was measured against, and any of those changing invalidates it. The three changes that most often silently move the optimum are adding an analyzed field (raises per-document cost, lowers the ideal payload), increasing shard count (raises achievable concurrency), and moving to instances with different disk throughput (moves the merge ceiling, which caps sustained rather than burst rate). Treat the sweep as a scheduled maintenance task tied to those events rather than a one-off, and keep the harness in the repository so re-running it costs an afternoon rather than a week of rediscovery.

Record the outcome where the next engineer will find it. A configuration file with a measured_on comment, as in step 4, is the minimum; a short entry in the runbook explaining why concurrency is 4 and not 8 prevents the well-meaning optimisation that doubles it during an incident. The most expensive part of tuning is not the measurement — it is doing the measurement three times because nobody wrote down the result.

Verification

Confirm the deployed job reproduces the measured rate within a reasonable margin:

# Sample the indexing rate over 60 seconds from the index stats counter.
before=$(curl -s 'localhost:9200/bench/_stats/indexing' | jq '._all.total.indexing.index_total')
sleep 60
after=$(curl -s 'localhost:9200/bench/_stats/indexing' | jq '._all.total.indexing.index_total')
echo "docs/sec: $(( (after - before) / 60 ))"
# Expected: within ~15% of the sweep value, e.g. 24000-28000 for a 28.4k peak

Confirm nothing is being rejected at the deployed setting — a single rejection means the operating point is too aggressive:

curl -s 'localhost:9200/_nodes/stats/thread_pool' \
  | jq '[.nodes[].thread_pool.write.rejected] | add'
# Expected: 0

Common Pitfalls

Operational notes

A sweep result is an input to capacity planning, not just to a config file. Once you know the sustained ceiling, every downstream question has an arithmetic answer: how long a full reindex takes, how much backlog an outage can accumulate before recovery exceeds the maintenance window, and how many nodes a doubling of the corpus will need. Write those three numbers into the runbook next to the configuration, because they are what an on-call engineer actually needs at three in the morning — not the payload size itself.

Finally, treat the sweep as a regression test for the ingestion path. Running it against a staging cluster after a mapping change, an engine upgrade, or an instance-type migration surfaces throughput regressions before they become production incidents, and the harness has already been written. A twenty-minute abbreviated sweep at three configurations is usually enough to detect a meaningful regression, which is cheap enough to run on every infrastructure change rather than once a year.

Related

• Bulk indexing throughput tuning — the parent guide covering the write path and the settings this sweep holds fixed.
• Handling 429 rejections with backpressure — what to do when the operating point is exceeded in production anyway.
• Sizing shards and replicas for a new index — shard count sets the parallelism ceiling any sweep can reach.