Bulk Loading Graphs with gremlin-python

Writing your own gremlin-python bulk loader is the moment you discover that raw insert speed is the easy part and correctness under partial failure is the whole job. This guide walks the exact procedure for loading a large dataset from a Python process — enabling batch-loading, sizing the id block, driving an idempotent batched loader with explicit transaction boundaries, and reindexing to parity afterward — so a coordinator timeout at row forty million does not leave you with a graph that counts correctly but traverses wrong. It sits under the Bulk Data Loading reference and narrows it to one concrete task: a single-writer, resumable Python loader you can run, verify, and roll back. The specific failure it prevents is a duplicated or torn import — the state where a retried batch inserts second copies of vertices or abandons a chunk of edges, and every downstream traversal inherits the corruption.

The batch-commit-retry loop of an idempotent gremlin-python bulk loader Input rows are split into fixed-size chunks. Each chunk enters a transaction opened with g.tx().begin(), the loader issues mergeV upsert mutations, then calls commit. On success the committed offset advances and the next chunk begins. On failure the transaction rolls back, an exponential backoff delay elapses, and the same chunk is retried from its persisted offset; because every mutation is an idempotent upsert, the replay converges instead of duplicating. After the final chunk commits, a reindex job rebuilds the mixed index from durable storage and a parity check compares graph counts against the source before the load is declared complete. Next chunk batch_size rows tx.begin() mergeV upserts commit committed? durable ack Reindex rebuild index Verify count parity rollback · backoff · retry same offset, converges yes · advance offset no last chunk
Each chunk commits as a unit; a failed commit rolls back and retries the same offset with backoff, and because every mutation is an idempotent upsert the replay converges. After the last chunk, a reindex rebuilds the index and a parity check gates completion.

Prerequisites

Confirm every item before running the loader. The most common cause of a “successful” load that fails verification is skipping the schema declaration and letting implicit schema creation race the writers.

  • JanusGraph 0.6.x or 1.0.x on a CQL backend (Cassandra 3.11+/4.x or ScyllaDB), reachable over the Gremlin Server WebSocket endpoint.
  • gremlinpython matching your server’s TinkerPop line — 3.5.x for JanusGraph 0.6, 3.6.x for 1.0. A version skew produces cryptic serialization errors mid-load.
  • The complete schema declared and committed — every vertex label, edge label, and property key the load will touch, plus a composite index on the natural key used for mergeV. Without that index, each upsert does a full scan and the load slows to a crawl.
  • A resolved connection pool. The loader’s single writer still needs socket headroom; confirm storage.cql.max-connections-per-host against the JanusGraph Connection Pool Tuning Guide.
  • A source count. Record the exact number of source rows before loading — it is the parity target the final verify step checks against.
  • A maintenance window if you will enable storage.batch-loading, since the setting is read at graph initialization and needs a Gremlin Server restart.

Step 1 — Enable batch-loading and declare the schema

Set the load-mode properties, restart the server, then declare the full schema in one management transaction so no writer ever triggers an implicit schema lock.

properties 
# janusgraph.properties — load session
storage.batch-loading=true
schema.default=none
storage.buffer-size=20480
storage.cql.batch-statement-size=20
storage.cql.write-consistency-level=LOCAL_ONE
groovy 
// Declare labels, keys, and the natural-key index up front.
mgmt = graph.openManagement()
person = mgmt.makeVertexLabel('person').make()
key = mgmt.makePropertyKey('key').dataType(String.class).make()
mgmt.makePropertyKey('name').dataType(String.class).make()
mgmt.buildIndex('personByKey', Vertex.class).addKey(key).unique().buildCompositeIndex()
mgmt.commit()

Verify: confirm batch-loading is active and the index exists and is enabled.

groovy 
mgmt = graph.openManagement()
println mgmt.get('storage.batch-loading')          // -> true
mgmt.printIndexes()                                 // personByKey must show ENABLED
mgmt.commit()

Step 2 — Size the id block for the load

The default id block forces an allocator round-trip every few thousand vertices, which throttles a large single-writer load. Raise ids.block-size so one block covers tens of seconds of inserts, and reserve the next block early.

properties 
# Sized for a single writer inserting a few million vertices.
ids.block-size=5000000
ids.renew-timeout=300000
ids.renew-percentage=0.3

Verify: after restart, drive a small warm-up batch and confirm no allocation stalls appear in the log.

bash 
# No IDBlockSizer / renew warnings during a 10k-row warm-up run
grep -E "IDBlock|renew|allocation" /var/log/janusgraph/janusgraph.log | tail -20

An empty or clean result means the block covers the warm-up without a renewal round-trip; repeated renew messages mean the block is still too small for the writer’s rate.

Step 3 — Run the idempotent batched loader

Drive the load from the loader below. It chunks the source, commits each chunk as a unit, upserts with mergeV so retries converge, and backs off on failure. Persist the committed offset so a crash resumes rather than restarts.

python 
import time
import json
import logging
from pathlib import Path
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

logging.basicConfig(level=logging.INFO, format="%(asctime)s [%(levelname)s] %(message)s")
CKPT = Path("load.ckpt")


def load(ws_endpoint, rows, batch_size=2000, max_retries=5):
    conn = DriverRemoteConnection(ws_endpoint, "g")
    g = traversal().withRemote(conn)
    start = int(CKPT.read_text()) if CKPT.exists() else 0
    it = iter(rows)
    for _ in range(start):              # skip already-committed rows on resume
        next(it, None)
    committed = start
    idx = start
    while chunk := list(islice(it, batch_size)):
        for attempt in range(1, max_retries + 1):
            tx = g.tx()
            gtx = tx.begin()
            try:
                for r in chunk:
                    (gtx.mergeV({T.label: "person", "key": r["key"]})
                        .option(__.property("name", r["name"]))
                        .iterate())
                tx.commit()             # durable unit; offset only advances after this
                committed += len(chunk)
                idx += len(chunk)
                CKPT.write_text(str(idx))
                break
            except Exception as e:
                tx.rollback()           # discard partial batch cleanly
                if attempt == max_retries:
                    logging.error("chunk at offset %d abandoned: %s", idx, e)
                    conn.close()
                    raise
                delay = min(2 ** attempt, 30)
                logging.warning("retry %d at offset %d in %ss: %s", attempt, idx, delay, e)
                time.sleep(delay)
    logging.info("committed=%d", committed)
    conn.close()
    return committed


# rows = (json.loads(l) for l in open("people.ndjson"))
# load("ws://gremlin:8182/gremlin", rows)

Verify: compare the committed vertex count against the source count.

groovy 
// In the Gremlin console
g.V().hasLabel('person').count()      // must equal the source row count from Prerequisites

Step 4 — Reindex and verify parity

The load wrote only to storage. Rebuild the mixed index from durable rows, enable it, then verify that a search-backed lookup returns the same element a storage-backed id lookup does.

groovy 
mgmt = graph.openManagement()
mgmt.updateIndex(mgmt.getGraphIndex('personByName'), SchemaAction.REINDEX).get()
mgmt.commit()
mgmt = graph.openManagement()
mgmt.updateIndex(mgmt.getGraphIndex('personByName'), SchemaAction.ENABLE_INDEX).get()
mgmt.commit()

Verify: the same predicate must return equal counts through the index and through a full scan.

groovy 
g.V().has('person', 'name', 'Ada').count()          // index-backed
g.V().hasLabel('person').filter{ it.get().value('name') == 'Ada' }.count()  // scan
// The two counts must match; a mismatch means the reindex has not reached parity.

Also confirm the index reports ENABLED and the reindex completed via mgmt.printIndexes().

Fallback and rollback procedures

Validate between steps rather than stacking them. Each failure has a defined recovery.

If Step 1 fails (schema lock or wrong load mode). If writers see SchemaViolationException or the load serializes, an implicit schema creation is racing. Confirm schema.default=none, re-declare any missing label or key, restart, and retry.

If Step 2 shows repeated renew warnings. The block is undersized for the writer’s rate. Raise ids.block-size further and restart; do not lower ids.renew-percentage to compensate, which only delays the stall.

If Step 3 abandons a chunk. The loader stops at the last persisted offset in load.ckpt. Fix the underlying cause — backend saturation, pool exhaustion, or a malformed source row — then rerun the same command; the checkpoint resumes from the abandoned offset and the idempotent upserts make the replay safe. To roll the whole load back, drop and recreate the keyspace before restarting rather than deleting vertices one by one, which generates tombstones.

If Step 4 fails parity. A count mismatch means the reindex has not finished or a shard failed mid-rebuild. Do not re-run the data load. Re-issue SchemaAction.REINDEX, and if it stalls, recover the index through Reindexing & Recovery — the storage rows are durable, so only the index needs rebuilding.