Parallel Bulk Import Tuning

Adding workers to a bulk load is the fastest way to either double your throughput or collapse the coordinator — and which one you get depends entirely on how you partition the input and cap concurrency. This guide is the procedure for parallelizing a JanusGraph import safely: splitting the source by key range so workers do not collide on the same partitions, sizing worker and pool concurrency to the coordinator’s real ceiling, tuning ids.block-size so parallel writers stop contending on id allocation, and throttling to the index bulk queue while you measure actual throughput. It sits under the Bulk Data Loading reference and extends the single-writer loader into a fleet. The failure it prevents is coordinator saturation — the state where more workers produce less committed throughput because every extra connection deepens the native-transport queue and lengthens the tail on every commit.

Partitioned parallel workers feeding a bounded connection pool into a coordinator The source dataset is split into disjoint key ranges, one per worker, so no two workers write the same storage partition and id-allocation contention is minimized. Each worker commits idempotent batches through a shared, bounded connection pool whose size is capped below the coordinator's native-transport capacity. The pool multiplexes all worker traffic into the CQL coordinator on the storage backend; because the pool is the real concurrency ceiling, adding workers beyond the pool size only lengthens the acquisition queue, not throughput. A throttle valve between the pool and the coordinator paces writes so the index bulk queue downstream never overruns. KEY-RANGE WORKERS worker [00–3f] worker [40–7f] worker [80–bf] worker [c0–ff] disjoint id blocks Bounded pool size ≤ coord cap socket socket socket real concurrency ceiling Throttle pace to index queue CQL coordinator native-transport queue Cassandra / ScyllaDB workers > pool size → longer acquire queue, not more committed throughput
Workers own disjoint key ranges and id blocks; they share a bounded pool sized below the coordinator's capacity, and a throttle paces writes to the index queue. Adding workers past the pool size deepens the acquire queue rather than raising throughput.

Prerequisites

Confirm each item before scaling out. Parallelism amplifies every misconfiguration in the single-writer path, so a load that is not clean at one writer will not become clean at eight.

  • A working single-writer load. Validate the idempotent batched loader from Bulk Loading Graphs with gremlin-python end-to-end first; parallelism is an optimization on top of a correct loader, not a substitute for one.
  • A partitionable key. The source must carry a key you can split into disjoint ranges — a hash prefix, a numeric id range, or a natural shard key — so each worker owns a non-overlapping slice.
  • Known coordinator capacity. nodetool tpstats headroom on MutationStage and the per-node native_transport_max_threads ceiling, so you can size total sockets below what the coordinators absorb.
  • A resolved connection pool baseline from Connection Pooling, because the parallel workers share that socket ceiling and it is the real concurrency limit.
  • Replica topology fixed. Reconcile Replication Strategies before scaling writers — replica count and consistency level set the coordinator fan-out each parallel commit multiplies.
  • A throughput baseline from the single-writer run (committed vertices per second) to measure the parallel speedup against.

Step 1 — Partition the input by key range

Split the source into disjoint ranges so no two workers write the same storage partition. Range partitioning on a hash prefix keeps each worker’s writes local to a slice of the token ring and eliminates cross-worker uniqueness collisions.

python 
def partition_ranges(num_workers: int):
    # Disjoint hex-prefix ranges over a byte of key space.
    step = 256 // num_workers
    for w in range(num_workers):
        lo = w * step
        hi = 256 if w == num_workers - 1 else (w + 1) * step
        yield (f"{lo:02x}", f"{hi:02x}")

def rows_for_range(source, lo, hi):
    for r in source:
        if lo <= r["key"][:2] < hi:      # key's hex prefix selects the owner
            yield r

Verify: confirm the ranges are disjoint and cover the whole key space with no gaps or overlaps.

python 
ranges = list(partition_ranges(4))
assert ranges[0][0] == "00" and ranges[-1][1] == "100"      # full coverage
assert all(ranges[i][1] == ranges[i+1][0] for i in range(len(ranges)-1))  # no overlap

Step 2 — Size worker and pool concurrency

The number of workers is not the throughput knob — the shared pool is. Size the pool below the coordinator’s native-transport capacity, then set worker count to keep the pool busy without deepening the acquire queue. A good starting point is workers equal to pool size, so each worker holds roughly one socket’s worth of in-flight commits.

properties 
# Shared across all parallel workers — this is the real ceiling.
storage.cql.core-connections-per-host=4
storage.cql.max-connections-per-host=16
storage.cql.max-requests-per-connection=1024
storage.cql.connection-timeout=5000

Total sockets must stay under coordinator capacity: with NN storage nodes and per-host max mm, the fleet opens up to N×mN \times m sockets, and that product must sit below aggregate native_transport_max_threads. Set worker count to the pool size and no higher — extra workers only queue on acquisition.

Verify: under a short canary run, in-flight sockets should approach the pool max while the coordinator keeps headroom.

bash 
# Coordinator not saturated: Active < max, Pending near zero
nodetool tpstats | grep -E "Pool Name|MutationStage|Native-Transport"

If MutationStage Pending climbs and All time blocked is non-zero, you have too many sockets for the coordinator — lower max-connections-per-host before adding workers.

Step 3 — Tune id-block-size against cross-worker contention

Parallel writers all draw ids from the same allocation mechanism, and if their blocks are small they contend on the id-allocation partition — the single most common reason parallel loads scale sublinearly. Give each worker a block large enough that renewals are rare, and let each worker’s disjoint key range keep its allocations independent.

properties 
# Each worker holds a large block; renewals become rare under parallelism.
ids.block-size=20000000
ids.renew-timeout=600000
ids.renew-percentage=0.3

Verify: allocation contention should not appear in the logs during the parallel run.

bash 
# No cross-worker renewal contention under full parallelism
grep -Ec "renew|IDBlock|Temporary.*id" /var/log/janusgraph/janusgraph.log

A near-zero count means blocks are large enough; a rising count means the writers are contending and ids.block-size must go higher.

Step 4 — Throttle to the index queue and measure throughput

If you dispatch to a live index, the index bulk queue is the binding constraint, not the storage coordinator. Throttle the aggregate commit rate to the queue’s drain rate, then measure committed throughput to confirm the parallel speedup is real. Aggregate throughput follows the pool ceiling, not the worker count:

Tagg=min⁡ ⁣(w×tw,  Pℓ)T_{\text{agg}} = \min\!\left(w \times t_w,\; \frac{P}{\ell}\right)

where ww is worker count, twt_w per-worker throughput, PP the pool socket count, and ℓ\ell the mean commit latency. Once w×tww \times t_w exceeds P/ℓP/\ell, more workers add nothing — the pool is saturated.

python 
import time
def measure(load_fn, workers, source_ranges):
    start = time.monotonic()
    total = sum(load_fn(r) for r in source_ranges)   # committed vertices
    elapsed = time.monotonic() - start
    print(f"workers={workers} committed={total} rate={total/elapsed:.0f}/s")

Verify: the measured rate must rise with workers up to the pool ceiling, then plateau. If it plateaus early or regresses, the pool or coordinator is the bottleneck, not worker count.

bash 
# Index queue must not reject during the throttled run
curl -s 'localhost:9200/_cat/thread_pool/write?v&h=name,queue,rejected'

rejected climbing means the throttle is too loose — lower the aggregate rate or defer the index and reindex after the load, per the Bulk Data Loading reference.

Fallback and rollback procedures

Change one variable at a time; parallel loads make compound changes impossible to attribute.

If Step 1 leaves gaps or overlaps. Overlapping ranges make two workers write the same key and race the uniqueness check; gaps silently drop rows. Re-derive the ranges from the assertion in Step 1 and re-run the coverage check before loading anything.

If Step 2 saturates the coordinator. MutationStage Pending climbing means total sockets exceed coordinator capacity. Lower max-connections-per-host and reduce worker count together — do not raise the pool to “push through,” which only deepens the queue. If you must absorb the load, add storage nodes rather than sockets.

If Step 3 still shows renewal contention. Raise ids.block-size further and confirm each worker’s key range is genuinely disjoint; overlapping ranges reintroduce contention no block size can fix.

If Step 4 rejects at the index. Stop dispatching during the load. Hold the index in REGISTERED state, complete the storage load, then reindex — a paced rebuild always beats fighting live backpressure. To roll back a corrupted parallel load, drop and recreate the keyspace rather than deleting per worker, then restart from clean state.