Neo4j Nodestream Plugin for Nodestream

This plugin provides a Nodestream interface to Neo4j.

Installation

pip install nodestream-plugin-neo4j

Usage

# nodestream.yaml
targets:
  neo4j:
    database: neo4j
    uri: bolt://localhost:7687
    username: neo4j
    password: neo4j
    database_name: neo4j # optional; name of the database to use.
    use_enterprise_features: false # optional; use enterprise features (e.g. node key constraints)

Extractor

The Neo4jExtractor class represents an extractor that reads records from a Neo4j database. It takes a single Cypher query as input and yields the records read from the database. The extractor will automatically paginate through the database until it reaches the end. Therefore, the query needs to include a SKIP and LIMIT clause. For example:

- implementation: nodestream_plugin_neo4j.extractor:Neo4jExtractor
  arguments:
    query: MATCH (p:Person) WHERE p.name = $name RETURN p.name SKIP $offset LIMIT $limit
    uri: bolt://localhost:7687
    username: neo4j
    password: neo4j
    database_name: my_database # Optional; defaults to neo4j
    limit: 100000 # Optional; defaults to 100
    parameters:
      # Optional; defaults to {}
      # Any parameters to be passed to the query
      # For example, if you want to pass a parameter called "name" with the value "John Doe", you would do this:
      name: John Doe

The extractor will automatically add the SKIP and LIMIT clauses to the query. The extractor will also automatically add the offset and limit parameters to the query. The extractor will start with offset set to 0 and limit set to 100 (unless overridden by setting limit) The extractor will continue to paginate through the database until the query returns no results.

Copy Command

The nodestream copy command copies nodes and relationships from one Neo4j target to another. The Neo4j plugin exposes the following --retriever-option parameters to tune throughput.

Retriever options (--retriever-option key=value)

Connector options (--connector-option key=value)

Best-known configuration (relationships, 24 h incremental)

Benchmark result: ~27 min, ~776 relationships/s peak for a production graph with ~1.27 M relationships (2026-06-24, sample_ratio=100).

nodestream copy \
  --from <source> \
  --to <destination> \
  --all \
  --retriever-option relationships_only=true \
  --retriever-option latest_hours=24 \
  --retriever-option shard_size=10000 \
  --concurrency-limit 20 \
  --flush-concurrency 8 \
  --batch-size 100000 \
  --step-outbox-size 2000000 \
  --reporting-frequency 30 \
  --connector-option execute_chunks_in_parallel=true \
  --connector-option chunk_size=10000 \
  --connector-option retries_per_chunk=5 \
  --metrics-interval-in-seconds 30 \
  --json

Why these values:

• shard_size=10000 — splits large relationship types into independent coroutines, filling all concurrency slots.
• concurrency-limit=20 — higher values (e.g. 30–40) cause SessionExpired/network errors on the destination; 20 gives 0–50 errors vs. 200+ at 30.
• flush-concurrency=8 — 8 parallel write lanes; saturates destination write throughput without overloading connections.
• batch-size=100000 — larger batches amortise per-batch overhead; peak throughput higher than 50K batches.
• step-outbox-size=2000000 — large buffer so the extractor is never blocked waiting for the writer to drain.
• relationships_only=true + latest_hours=24 — incremental mode; nodes are stable, only new/changed relationships need copying.

Concepts

Migrations

The plugin supports migrations. Migrations are used to create indexes and constraints on the database.

As part of the migration process, the plugin will create __NodestreamMigration__ nodes in the database. This node will have a name property that is set to the name of the migration.

Additionally, the plugin will create a __NodestreamMigrationLock__ node in the database. This node will be exit when the migration process is running and will be deleted when the migration process is complete. This is used to prevent multiple migration processes from running at the same time.