Bulk Data Loading

Bulk ingestion is where a JanusGraph cluster that behaves perfectly under interactive traffic quietly destroys itself. The moment you replay a hundred million rows through the same transactional path that serves single-vertex lookups, three subsystems fail at once: the id allocator hands out blocks faster than it can renew them, the mixed-index backend drowns under a bulk queue it was never asked to throttle, and half-committed batches leave orphaned edges pointing at vertices that no longer exist. This reference sits under the JanusGraph Storage Backend Architecture & Configuration reference and isolates one operation: moving a large, bounded dataset into the graph without corrupting state or saturating the index. The failure mode it prevents is a torn load — a partially applied import that passes a naive vertex count but fails every traversal that crosses a batch boundary. Everything below prioritizes idempotent, bounded, resumable ingestion over the raw insert throughput a driver benchmark rewards.

The diagram below shows the shape of a correct bulk load: writes stream into storage with the index dispatch deferred, then a reindex job rebuilds the mixed index from the durable storage rows once the load completes.

Core Configuration & Consistency Tuning

JanusGraph offers two distinct load modes, and choosing the wrong one is the first mistake. The batch-loading mode trades consistency checks for throughput: it disables locking, relaxes uniqueness enforcement, and assumes the caller guarantees data cleanliness. The standard transactional mode keeps every check live and is the only safe choice when the incoming data can violate a uniqueness constraint or references vertices that may not exist. Use batch-loading only when the dataset is pre-deduplicated and internally consistent; use the transactional path with idempotent upserts when it is not.

The batch-loading profile in janusgraph.properties:

storage.backend=cql
storage.hostname=10.0.1.10,10.0.1.11,10.0.1.12
storage.cql.keyspace=janusgraph_prod

# --- Load mode ---
# Disables locking and consistency checks for the load session.
storage.batch-loading=true
# Skip implicit schema creation; declare the full schema up front instead.
schema.default=none

# --- Id block allocation ---
# Larger blocks = fewer allocator round-trips, but more ids wasted on crash.
ids.block-size=10000000
# How long a partition waits for a contested block before failing.
ids.renew-timeout=600000
# Renew the next block before the current one is exhausted.
ids.renew-percentage=0.3

# --- Write buffering ---
# Mutations buffered before a backend flush; raise it for sequential loads.
storage.buffer-size=20480
# CQL statements batched into one coordinator round-trip.
storage.cql.batch-statement-size=20

# --- Consistency during load ---
storage.cql.write-consistency-level=LOCAL_ONE
storage.cql.read-consistency-level=LOCAL_QUORUM

Operational constraints, in the order they bite during a large import:

1. storage.batch-loading=true removes your safety net, so the loader must replace it. With locking disabled, JanusGraph will not stop you from writing two vertices that violate a uniqueness index — it will happily persist both and corrupt the constraint. Only enable it when the loader itself guarantees each key is written exactly once, or when you pair it with mergeV upserts that converge on replay.
2. schema.default=none forces every label and property key to exist before the load starts. Implicit schema creation (schema.default=default) takes a schema lock on first sight of an unknown label, and under concurrent bulk writers that lock serializes the whole import. Declare the complete schema in a single management transaction first, then load against a frozen schema.
3. ids.block-size is the single highest-leverage throughput knob. JanusGraph assigns each vertex a globally unique id from a block reserved through the storage backend. A block that is too small forces an allocator round-trip every few thousand vertices, and under many parallel writers those round-trips contend on the same id-allocation partition. Size the block so each writer holds enough ids for tens of seconds of work — millions for a large load — but no larger, because every id above the high-water mark at crash time is permanently wasted.
4. ids.renew-timeout decides how long a stalled writer waits before it fails. If a block is contested and cannot be renewed within this window, the transaction throws. Raise it for wide-parallel loads where allocation contention is expected; leave it default for a single-threaded import.
5. storage.buffer-size and storage.cql.batch-statement-size govern how many mutations coalesce into one backend round-trip. A larger buffer amortizes network latency across more writes, which is exactly what a sequential bulk load wants — but it also widens the blast radius of a failed flush, so keep it bounded and let the application-level batch (below) be the real unit of commit.

The correct id-block sizing follows directly from writer count and ingest rate. If $r$ is the aggregate vertex insert rate and $w$ the number of parallel writers, a block should cover at least a renewal interval $t$ of work per writer:

$$\text{ids.block-size} \geq \frac{r \times t}{w}$$

These settings interact with pool sizing and replica topology. Before a large load, reconcile max-connections-per-host through Connection Pooling so the batched writers do not starve the pool, and confirm the keyspace, compaction, and consistency baseline through Cassandra Backend Setup — a load against an under-provisioned keyspace triggers tombstone accumulation and compaction storms that outlast the import itself.

Index Synchronization: Deferring Dispatch and Reindexing

The mixed index is the component most likely to fail a bulk load, because every mutation that touches an indexed property normally dispatches a corresponding document to Elasticsearch or OpenSearch. At interactive rates that dispatch is invisible; at bulk rates it becomes a firehose that overruns the index bulk queue, and once the queue rejects, the index falls behind the storage commit and the graph desynchronizes. The discipline that prevents this is to keep the index out of the write path entirely during the load, then rebuild it from durable storage afterward.

There are two ways to defer dispatch. The first is to register the mixed index in ENABLED state but load through the batch-loading session, which keeps the storage write synchronous while the index dispatch is queued and throttled rather than blocking the commit. The second, and the one that scales to billions of elements, is to create the index but hold it in INSTALLED/REGISTERED state — never transitioning it to ENABLED — so no dispatch occurs at all during the load, then run a reindex job that reads committed storage rows and populates the index in one controlled pass:

// After the bulk load completes and id blocks have drained,
// build the mixed index from durable storage state.
mgmt = graph.openManagement()
idx = mgmt.getGraphIndex('vByEmail')
mgmt.updateIndex(idx, SchemaAction.REINDEX).get()
mgmt.commit()

// Then transition the index to usable state.
mgmt = graph.openManagement()
mgmt.updateIndex(mgmt.getGraphIndex('vByEmail'), SchemaAction.ENABLE_INDEX).get()
mgmt.commit()

Reindexing from storage is strictly safer than live dispatch during load because it derives the index from committed rows, not from an in-flight mutation stream that a dropped connection can reorder. The bulk queue is sized and paced by the reindex job, not by your ingest rate. The full acknowledgment-boundary trade-off — how long to wait before a post-load read is considered index-consistent — is covered in Eventual vs Strong Consistency, and the routing that decides which backing index answers a query lives in Mixed-Index Routing. When a reindex stalls or a shard fails partway through the rebuild, recover it through Reindexing & Recovery rather than restarting the whole load. Backend-specific propagation behavior is in the Elasticsearch integration and OpenSearch sync patterns guides.

Poll the index bulk queue rather than assuming the reindex keeps pace. Elasticsearch /_cat/thread_pool/write?v bulk queue depth and the JanusGraph index.[name].elasticsearch backpressure counters are the first signals to move; if the queue depth climbs toward its bound during the reindex, throttle the reindex batch size before it starts rejecting.

Python Integration Pattern

The loader is where idempotency is won or lost. A bulk import will fail partway through — a coordinator times out, a pool socket drops, a block renewal stalls — and the only safe response is a retry that converges rather than duplicates. That means every mutation must be a mergeV/coalesce upsert keyed on a natural identity, never a bare addV, and every batch must commit or roll back as a unit. The following gremlin-python loader chunks the input, commits in bounded batches with explicit transaction boundaries, and backs off on failure.

import time
import logging
from itertools import islice
from gremlin_python.driver.driver_remote_connection import DriverRemoteConnection
from gremlin_python.process.anonymous_traversal import traversal
from gremlin_python.process.graph_traversal import __
from gremlin_python.process.traversal import T

logger = logging.getLogger(__name__)


class BulkLoader:
    def __init__(self, ws_endpoint: str, batch_size: int = 2000,
                 max_retries: int = 5):
        # batch_size is the commit unit. Keep it in the 500-10000 range:
        # too small multiplies id-block and commit overhead, too large
        # widens the rollback blast radius and holds pool slots longer.
        self.conn = DriverRemoteConnection(ws_endpoint, "g")
        self.g = traversal().withRemote(self.conn)
        self.batch_size = batch_size
        self.max_retries = max_retries

    def _chunks(self, rows):
        it = iter(rows)
        while chunk := list(islice(it, self.batch_size)):
            yield chunk

    def _commit_batch(self, tx_g, chunk):
        # Idempotent by construction: mergeV upserts on the natural key,
        # so a replayed retry converges instead of duplicating vertices.
        for row in chunk:
            (tx_g.mergeV({T.label: row["label"], "key": row["key"]})
                 .option(__.property("name", row["name"])
                           .property("payload", row["payload"]))
                 .iterate())

    def load(self, rows):
        committed = 0
        for i, chunk in enumerate(self._chunks(rows)):
            for attempt in range(1, self.max_retries + 1):
                tx = self.g.tx()
                tx_g = tx.begin()
                try:
                    self._commit_batch(tx_g, chunk)
                    tx.commit()                 # unit of durability
                    committed += len(chunk)
                    break
                except Exception as e:
                    tx.rollback()               # release slot, discard partial batch
                    if attempt == self.max_retries:
                        logger.error("batch %d abandoned after %d attempts: %s",
                                     i, attempt, e)
                        raise
                    backoff = min(2 ** attempt, 30)
                    logger.warning("batch %d retry %d in %ss: %s",
                                   i, attempt, backoff, e)
                    time.sleep(backoff)         # exponential backoff, capped
            if i % 50 == 0:
                logger.info("progress: %d vertices committed", committed)
        logger.info("load complete: %d vertices committed", committed)
        return committed

    def close(self):
        self.conn.close()

Implementation requirements:

• mergeV keyed on a natural identity is the whole safety model. If the retry replays a batch that was already partially applied, the upsert matches the existing vertex and updates it in place. A bare addV would insert a duplicate on every retry, and duplicates in batch-loading mode silently corrupt any uniqueness index.
• Commit is the batch boundary, not the row boundary. Committing every row multiplies coordinator round-trips and id-block pressure; committing the whole load in one transaction blows the heap and makes any failure a total loss. The 500–10000 range keeps both bounded.
• Rollback on failure discards the partial batch cleanly so the retry starts from a known state. Never catch and continue without rolling back — a half-applied transaction leaves the pool slot held and the batch in an ambiguous state.
• Exponential backoff spreads reconnection attempts across a recovering backend node instead of hammering it. Cap it so a persistent failure surfaces as an abandoned batch you can requeue, not an infinite loop.

For the mapping from this batch concurrency to pool and JVM thread limits, work through the JanusGraph Connection Pool Tuning Guide.

Commit Cadence & Batch Management

A bulk load is a sequence of small transactions, not one large one, and the size of each transaction is the control variable that determines whether the load completes or thrashes. Treat the batch as a resource that holds an id-block reservation, a pool slot, and a write buffer for its entire lifetime.

Batch sizing rules:

• Size the batch from commit latency, not row count intuition. A batch that takes longer than a few seconds to commit is either too large or aimed at a saturated backend; a batch that commits in milliseconds is too small and is paying fixed transaction overhead on every flush. The 500–10000 window brackets the range where commit overhead is amortized but heap and rollback cost stay bounded.
• Keep batch size below the id-block renewal interval. If a single batch consumes more ids than a block holds, the loader stalls mid-batch waiting on a renewal round-trip, and that stall counts against ids.renew-timeout. Size ids.block-size so at least several batches fit inside one block.
• Throttle between batches under index pressure. When you are dispatching to a live index rather than deferring, insert a short sleep between commits so the index bulk queue drains. Throughput you cannot index is throughput you will lose to backpressure.

Resumability rules:

• Checkpoint the committed offset, not the wall clock. Because every mutation is an idempotent upsert, a resumed load can safely replay the last in-flight batch — but only if you know which offset last committed. Persist the batch index alongside the load so a restart resumes from the last durable checkpoint.
• Drain id blocks before reindexing. A reindex reads committed storage rows; if writers still hold un-flushed id blocks, the reindex sees a partial graph. Confirm all writers have committed and closed before transitioning the index to ENABLED.

Diagnostics & Operational Fallbacks

Expose id-allocation, storage-buffer, and index-queue metrics through JMX, then triage against these symptom/diagnosis/resolution triplets.

• Symptom: the load stalls periodically, and logs show IDBlockSizer or Temporary allocation exceptions. Diagnosis: id-block exhaustion under parallel writers — blocks are consumed faster than they renew, and contention on the id-allocation partition serializes the writers. Resolution: raise ids.block-size so each writer holds tens of seconds of work, raise ids.renew-timeout to survive contention, and lower ids.renew-percentage so the next block is reserved earlier, before the current one drains.

• Symptom: read latency climbs across the whole graph during and after the load, and nodetool tablestats shows rising tombstone counts. Diagnosis: tombstone accumulation — overwrites and deletes during the load (or a load that writes then deletes on retry without idempotent upserts) generate tombstones that compaction cannot keep pace with. Resolution: switch retries to mergeV upserts so replays overwrite in place instead of delete-then-insert, tune the keyspace to LeveledCompactionStrategy for write-heavy loads, and run a major compaction after the load drains the tombstones.

• Symptom: storage commits succeed but search queries return stale or missing results, and /_cat/thread_pool/write shows a growing bulk queue with rejections. Diagnosis: index backpressure — the mixed index cannot absorb the dispatch rate and is rejecting bulk requests, so the index falls behind committed storage. Resolution: stop dispatching during the load; hold the index in REGISTERED state, load to storage only, then rebuild with a paced SchemaAction.REINDEX whose batch size you throttle to the bulk queue’s drain rate.

• Symptom: traversals that cross a batch boundary fail with missing-vertex errors even though vertex counts look correct. Diagnosis: a torn load — a batch committed its vertices but a later batch of edges referencing them was rolled back and never retried, leaving dangling references. Resolution: load vertices to completion before loading edges, verify the vertex checkpoint before starting edges, and requeue any abandoned batch from its persisted offset rather than declaring the load done.

• Symptom: NoHostAvailableException bursts during the load while nodetool status shows all nodes UN. Diagnosis: pool exhaustion under the batched writers, not node death — the concurrency of the load exceeds the connection pool ceiling. Resolution: reconcile the loader’s writer count with max-connections-per-host, lower the write consistency to LOCAL_ONE for the load session, and shorten connection-timeout so backpressure reaches the loader instead of queuing threads.

For quorum baselines behind the consistency settings, see the upstream Apache Cassandra consistency levels reference.

Related

• Up a level: JanusGraph Storage Backend Architecture & Configuration — the storage tier this load writes into.
• Bulk Loading Graphs with gremlin-python — the step-by-step procedure for the idempotent batched loader above.
• Parallel Bulk Import Tuning — partitioning input and sizing worker concurrency without overloading the coordinator.
• Connection Pooling — the socket ceiling the batched writers share.
• Cassandra Backend Setup — keyspace, compaction, and consistency the load depends on.
• Reindexing & Recovery — rebuilding and recovering the mixed index after the load completes.