How do I load a large graph into an empty database?

You have a large input and an empty database, and loading it the ordinary way, one batch at a time, is too slow. Each batch asks the database, for every vertex and edge, whether it already exists.

A bulk load answers that question once, before writing anything. GraFlo casts the whole input to a staging directory and folds every vertex and edge to one row by its identity. It then writes each row once, with no lookups. The graph it ends with is the one an ordinary load of the same input would produce.

flowchart LR
    csv[CSV file] -- cast --> stage[(artifacts/staging)]
    stage -- fold --> stage
    stage -- write once --> neo4j[(Neo4j)]

What you need

• GraFlo with the bulk extra (pip install 'graflo[bulk]').
• The repository's Neo4j container, see docker/README.md.

The data

data/works.csv lists papers with their author and venue. Paper W2 appears twice: the second row corrects a typo in its title. manifest.yaml makes a work vertex per paper, an author and a venue, and an edge from each work to its author and to its venue.

Steps

1. Load in bulk

ingest.py asks for bulk mode and names the staging directory:

engine.define_and_ingest(
    manifest=manifest,
    target_db_config=conn_conf,
    ingestion_params=IngestionParams(mode="bulk", staging_dir=staging),
    recreate_schema=True,
)

cd examples/13-bulk-load
uv run python ingest.py

From the command line, the same load is uv run ingest --db-config-path db.yaml --schema-path manifest.yaml --bulk --staging-dir artifacts/staging.

2. See what the fold did

The script prints artifacts/staging/counts.json: five work rows were staged and four folded, since both rows of W2 became one vertex holding the later title. Ada wrote two papers, so she is one author; the two written_by rows of W2 are one edge. The load then checked that Neo4j holds exactly these counts.

3. Resume a stopped load

Stop the script during the write and run it again with the same staging directory. It does not cast or fold again, skips the chunks it wrote, and writes again the one chunk whose commit it did not record. To start over, delete artifacts/staging.

What next

Bulk load lists what bulk mode cannot load, which databases support it, and how to tune it.

Files

The example lives in examples/13-bulk-load.

manifest.yaml

schema:
    metadata:
        name: works
    graph:
        vertex_config:
            vertices:
            -   name: work
                properties:
                -   id
                -   title
                -   year
                identity:
                -   id
            -   name: author
                properties:
                -   name
                identity:
                -   name
            -   name: venue
                properties:
                -   name
                identity:
                -   name
        edge_config:
            edges:
            -   source: work
                target: author
                relation: written_by
            -   source: work
                target: venue
                relation: published_in
    db_profile: {}
ingestion_model:
    resources:
    -   name: works
        pipeline:
        -   vertex: work
            from:
                id: id
                title: title
                year: year
        -   vertex: author
            from:
                name: author
        -   vertex: venue
            from:
                name: venue
bindings:
    connectors:
    -   regex: "^works.*\\.csv$"
        sub_path: data
        resource_name: works

ingest.py

"""How do I load a large graph into an empty database?

Reads ``manifest.yaml`` and loads ``data/works.csv`` into Neo4j in bulk: the
input is cast and staged in ``artifacts/staging``, folded to one row per vertex
and edge, then written once. Run it from this directory:

    uv run python ingest.py

Run it again to resume a stopped load; delete ``artifacts/staging`` to start over.
"""

import json
from pathlib import Path

from suthing import FileHandle

from graflo import GraphManifest
from graflo.connections import Neo4jConfig
from graflo.hq import GraphEngine
from graflo.hq.caster import IngestionParams

manifest = GraphManifest.from_config(FileHandle.load("manifest.yaml"))
manifest.finish_init()

# Connection settings of the Neo4j container started from docker/neo4j.
conn_conf = Neo4jConfig.from_docker_env()
staging = Path("artifacts/staging")

engine = GraphEngine(target_db_flavor=conn_conf.connection_type)
engine.define_and_ingest(
    manifest=manifest,
    target_db_config=conn_conf,
    ingestion_params=IngestionParams(mode="bulk", staging_dir=staging),
    # The target must be empty: drop whatever an earlier run left there.
    recreate_schema=True,
)

for run in json.loads((staging / "counts.json").read_text())["runs"]:
    kind = run["type"]
    name = kind["name"] if kind["kind"] == "vertex" else " -> ".join(kind["edge_id"])
    print(f"{name}: {run['staged']} staged, {run['folded']} folded")