name	neo4j-spark-skill
description	Use when reading from or writing to Neo4j with Apache Spark or Databricks using the
version	1.0.3
allowed-tools	Bash WebFetch

Neo4j Connector for Apache Spark

When to Use

• Reading Neo4j nodes/relationships into Spark DataFrames
• Writing Spark DataFrames to Neo4j as nodes or relationships
• Databricks notebooks connecting to Neo4j
• Delta Lake → Neo4j ingestion pipelines
• Partitioned parallel reads from large Neo4j graphs

When NOT to Use

• Python bolt driver / execute_query → neo4j-driver-python-skill
• Cypher query writing → neo4j-cypher-skill
• GDS graph algorithms → neo4j-gds-skill
• Spring Boot + Neo4j → neo4j-spring-data-skill

---

Version Matrix

Connector	Spark	Scala	Java	Databricks Runtime	Neo4j	Maven coordinate
6.0.x	4.0, 4.1	2.13	17+	17.3 LTS	5.x, 2025.x, 2026.x	org.neo4j.connectors:spark:6.0.0-s_2.13
5.5.x / 5.4.x	3.4, 3.5	2.12, 2.13	8+	14.3–16.4 LTS	4.4, 5.x, 2025.x, 2026.x	org.neo4j:neo4j-connector-apache-spark_2.13:5.5.0_for_spark_3

Group ID changed in 6.0 — org.neo4j:neo4j-connector-apache-spark_<scala> is now a relocation POM pointing at org.neo4j.connectors:spark. On Spark 3.x stay on 5.5.x.

6.0 breaking changes

Change	Migration
Spark baseline 3.5 → 4.0/4.1; Scala 2.12 and Java 8–11 dropped	Upgrade to 5.5.0 first, then Spark 4.x + Scala 2.13 + Java 17
Maven coordinate org.neo4j.connectors:spark:<version>-s_2.13	Replace old _for_spark_3 coordinate
schema.optimization.type removed	schema.optimization.node.keys, schema.optimization.relationship.keys, schema.optimization
$stream.offset in partitioned reads removed	Use partitions + query.count
;-separated multi-statement script removed	script.1, script.2, … script.N — executed in numbered order
relationship.save.strategy default native → keys	Set .option("relationship.save.strategy", "native") explicitly to keep old behaviour
query option rewritten for Data Source V2 predicate push-down	No action; verify plans on upgrade

---

Setup

Standalone Spark (PySpark)

from pyspark.sql import SparkSession

spark = (SparkSession.builder
    .appName("neo4j-app")
    .config("spark.jars.packages",
            "org.neo4j.connectors:spark:6.0.0-s_2.13")   # Spark 3.x: org.neo4j:neo4j-connector-apache-spark_2.13:5.5.0_for_spark_3
    .config("neo4j.url", "neo4j+s://xxxx.databases.neo4j.io")
    .config("neo4j.authentication.type", "basic")
    .config("neo4j.authentication.basic.username", "neo4j")
    .config("neo4j.authentication.basic.password", "password")
    .getOrCreate())

Standalone Spark (Scala)

val spark = SparkSession.builder
  .appName("neo4j-app")
  .config("spark.jars.packages",
    "org.neo4j.connectors:spark:6.0.0-s_2.13")
  .config("neo4j.url", "neo4j+s://xxxx.databases.neo4j.io")
  .config("neo4j.authentication.type", "basic")
  .config("neo4j.authentication.basic.username", "neo4j")
  .config("neo4j.authentication.basic.password", "password")
  .getOrCreate()

Databricks — Cluster Installation

1. Cluster → Libraries → Install New → Maven
2. Coordinate org.neo4j.connectors:spark:6.0.0-s_2.13 on DBR 17.3 LTS; org.neo4j:neo4j-connector-apache-spark_2.13:5.5.0_for_spark_3 on DBR 14.3–16.4 LTS
3. Cluster → Advanced Options → Spark tab — add config:

   neo4j.url neo4j+s://xxxx.databases.neo4j.io
   neo4j.authentication.type basic
   neo4j.authentication.basic.username {{secrets/neo4j/username}}
   neo4j.authentication.basic.password {{secrets/neo4j/password}}
   

4. Use Single user access mode (Unity Catalog shared mode not supported)

Databricks — Secrets (preferred over plaintext)

# Store credentials once:
# databricks secrets create-scope --scope neo4j
# databricks secrets put --scope neo4j --key url
# databricks secrets put --scope neo4j --key username
# databricks secrets put --scope neo4j --key password

neo4j_url  = dbutils.secrets.get(scope="neo4j", key="url")
neo4j_user = dbutils.secrets.get(scope="neo4j", key="username")
neo4j_pass = dbutils.secrets.get(scope="neo4j", key="password")

spark.conf.set("neo4j.url", neo4j_url)
spark.conf.set("neo4j.authentication.type", "basic")
spark.conf.set("neo4j.authentication.basic.username", neo4j_user)
spark.conf.set("neo4j.authentication.basic.password", neo4j_pass)

---

Key Configuration Options

Option	Description	Default
neo4j.url	Bolt/Neo4j URI	— (required)
neo4j.authentication.type	none, basic, kerberos, bearer	basic
neo4j.authentication.basic.username	Username	driver default
neo4j.authentication.basic.password	Password	driver default
neo4j.authentication.bearer.token	Bearer token	—
neo4j.database	Target database	driver default
neo4j.access.mode	read or write	read
neo4j.encryption.enabled	TLS (ignored with +s/+ssc URI)	false
neo4j.db.transaction.timeout	Transaction timeout (ms)	driver default
neo4j.db.transaction.metadata.<key>	Custom transaction metadata surfaced in query log [6.0]	empty
neo4j.authentication.type = supplier name	Custom AuthenticationTokenSupplierFactory (e.g. keycloak via org.neo4j.connectors:commons-authn-keycloak) for expiring OAuth/OIDC tokens	—

Cypher version and query tuning [6.0]

Option	Effect
cypher.version	Cypher language version — 5 (default) or 25
cypher.tuning.<param>	Emits CYPHER <param>=<value> preamble on every generated query

Valid with labels, relationship, query on reads and writes; rejected with gds.

df = (spark.read.format("org.neo4j.spark.DataSource")
    .option("query", "MATCH (o:Object) RETURN o.id AS id, o.name AS name")
    .option("cypher.version", "25")
    .option("cypher.tuning.runtime", "parallel")           # CYPHER 25 runtime=parallel
    .option("db.transaction.metadata.app", "spark-etl")    # tags transactions in query.log
    .load())

---

Reading from Neo4j

Three mutually exclusive read modes — use exactly one per .read() call.

Label scan (nodes)

# PySpark
df = (spark.read.format("org.neo4j.spark.DataSource")
    .option("labels", ":Person")
    .load())
df.printSchema()
df.show()

// Scala
val df = spark.read
  .format("org.neo4j.spark.DataSource")
  .option("labels", ":Person")
  .load()

Multi-label filter (AND): .option("labels", ":Person:Employee")

Result includes <id> (internal Neo4j id) and <labels> columns.

Cypher query read

df = (spark.read.format("org.neo4j.spark.DataSource")
    .option("query", "MATCH (p:Person)-[:ACTED_IN]->(m:Movie) RETURN p.name AS actor, m.title AS movie, m.year AS year")
    .load())

Use explicit RETURN aliases — they become DataFrame column names. No SKIP/LIMIT in query (connector handles pagination).

Relationship scan

df = (spark.read.format("org.neo4j.spark.DataSource")
    .option("relationship", "BOUGHT")
    .option("relationship.source.labels", ":Customer")
    .option("relationship.target.labels", ":Product")
    .load())

Result columns: <rel.id>, <rel.type>, <source.*>, <target.*>, plus relationship properties.

Read partition tuning

df = (spark.read.format("org.neo4j.spark.DataSource")
    .option("labels", ":Transaction")
    .option("partitions", "10")        # parallel partitions (default: 1)
    .option("batch.size", "5000")      # rows per partition batch (default: 5000)
    .option("schema.flatten.limit", "100")  # rows sampled for schema inference
    .load())

Full read options reference: references/read-patterns.md

---

Writing to Neo4j

SaveMode

SaveMode	Cypher	Requires
Append	CREATE	nothing extra
Overwrite	MERGE	node.keys (nodes) or *.node.keys (rels)
ErrorIfExists	CREATE + error if exists	—

Always create uniqueness constraints on node.keys properties before writing in Overwrite mode.

Write nodes — Append (CREATE)

from pyspark.sql import Row

people = spark.createDataFrame([
    {"name": "Alice", "age": 30},
    {"name": "Bob",   "age": 25},
])

(people.write.format("org.neo4j.spark.DataSource")
    .mode("Append")
    .option("labels", ":Person")
    .save())

Write nodes — Overwrite (MERGE)

(people.write.format("org.neo4j.spark.DataSource")
    .mode("Overwrite")
    .option("labels", ":Person")
    .option("node.keys", "name")       # comma-separated; df_col:node_prop if names differ
    .save())

node.keys with rename: .option("node.keys", "df_col:node_property,id:personId")

Write nodes — Scala

import org.apache.spark.sql.SaveMode

peopleDF.write
  .format("org.neo4j.spark.DataSource")
  .mode(SaveMode.Overwrite)
  .option("labels", ":Person")
  .option("node.keys", "name")
  .save()

Write relationships

Use coalesce(1) before relationship writes to avoid deadlocks.

rel_df = spark.createDataFrame([
    {"cust_id": "C1", "prod_id": "P1", "qty": 3},
    {"cust_id": "C2", "prod_id": "P2", "qty": 1},
])

(rel_df.coalesce(1)
    .write.format("org.neo4j.spark.DataSource")
    .mode("Append")
    .option("relationship", "BOUGHT")
    .option("relationship.save.strategy", "keys")
    .option("relationship.source.labels", ":Customer")
    .option("relationship.source.save.mode", "Match")          # require existing nodes
    .option("relationship.source.node.keys", "cust_id:id")
    .option("relationship.target.labels", ":Product")
    .option("relationship.target.save.mode", "Match")
    .option("relationship.target.node.keys", "prod_id:id")
    .option("relationship.properties", "qty:quantity")
    .save())

relationship.source.save.mode / relationship.target.save.mode:

• Match — find existing nodes (fail if missing)
• Append — always CREATE new nodes
• Overwrite — MERGE nodes

Pre-write scripts [6.0]

script.N runs Cypher once before write operations, in numbered order. Required for index/constraint setup when using query mode (schema.optimization.* rejected there).

(df.write.format("org.neo4j.spark.DataSource")
    .mode("Overwrite")
    .option("query", "MERGE (p:Person {email: event.email}) SET p.name = event.name")
    .option("script.1", "CREATE CONSTRAINT person_email IF NOT EXISTS FOR (p:Person) REQUIRE p.email IS UNIQUE")
    .option("script.2", "CREATE INDEX person_name IF NOT EXISTS FOR (p:Person) ON (p.name)")
    .option("index.await.timeout", "300")   # db.awaitIndexes seconds; 0 disables
    .save())

script (single statement) and script.N are mutually exclusive. Semicolon-separated statements inside one script fail on 6.0.

Full write options reference: references/write-patterns.md

---

Databricks — Delta Lake → Neo4j Pipeline

# Read from Delta table (Unity Catalog or DBFS)
delta_df = spark.read.format("delta").table("catalog.schema.customers")

# Optional: filter/transform in Spark before writing
filtered = delta_df.filter("active = true").select("customer_id", "name", "region")

# Write to Neo4j
(filtered.write.format("org.neo4j.spark.DataSource")
    .mode("Overwrite")
    .option("labels", ":Customer")
    .option("node.keys", "customer_id")
    .option("batch.size", "20000")
    .save())

Pipeline pattern for relationships — load both node sets first, then write edges:

# Step 1: ensure nodes exist
customers_df.write.format("org.neo4j.spark.DataSource").mode("Overwrite") \
    .option("labels", ":Customer").option("node.keys", "customer_id").save()

products_df.write.format("org.neo4j.spark.DataSource").mode("Overwrite") \
    .option("labels", ":Product").option("node.keys", "product_id").save()

# Step 2: write relationships (single partition)
orders_df.coalesce(1).write.format("org.neo4j.spark.DataSource").mode("Append") \
    .option("relationship", "ORDERED") \
    .option("relationship.save.strategy", "keys") \
    .option("relationship.source.labels", ":Customer") \
    .option("relationship.source.save.mode", "Match") \
    .option("relationship.source.node.keys", "customer_id:customer_id") \
    .option("relationship.target.labels", ":Product") \
    .option("relationship.target.save.mode", "Match") \
    .option("relationship.target.node.keys", "product_id:product_id") \
    .save()

---

Write Performance Tuning

Scenario	Recommendation
Node writes (no lock contention)	repartition(N) where N ≤ Neo4j CPU cores
Relationship writes (lock risk)	coalesce(1) — single partition
Large datasets	batch.size 10000–20000 (adjust to heap)
MERGE-heavy loads	Add uniqueness constraint on node.keys properties first

# Aggressive batch — monitor Neo4j heap; OOM risk above 50k
(big_df.repartition(8)
    .write.format("org.neo4j.spark.DataSource")
    .mode("Overwrite")
    .option("labels", ":Event")
    .option("node.keys", "event_id")
    .option("batch.size", "20000")
    .save())

---

Common Errors

Error	Cause	Fix
ClassNotFoundException: org.neo4j.spark.DataSource	JAR not on classpath	Add spark.jars.packages or attach library
Deadlock on relationship write	Multiple partitions locking nodes	coalesce(1) before write
Duplicate nodes on Overwrite	No uniqueness constraint on keys	CREATE CONSTRAINT ON (n:Label) ASSERT n.prop IS UNIQUE
OOM on Neo4j side	batch.size too large	Reduce to 5000–10000; check heap
Schema all string columns	No APOC, schema not sampled	Set schema.flatten.limit higher; or use query mode with explicit types
Access mode is read error on write	Session opened in read mode	Remove neo4j.access.mode or set to write
Databricks Shared cluster fails	Unity Catalog shared mode unsupported	Switch to Single User access mode
NoSuchMethodError / IncompatibleClassChangeError on Spark 4	5.x connector on a Spark 4 runtime	Use org.neo4j.connectors:spark:6.0.0-s_2.13
Relationship write ignores rel.* / source.* columns after upgrade	6.0 default strategy is keys, not native	.option("relationship.save.strategy", "native")
script option rejected with multiple statements	6.0 removed ;-separated scripts	Split into script.1, script.2, …

---

Checklist

• Connector coordinate matches Spark line — org.neo4j.connectors:spark:*-s_2.13 for Spark 4.x, org.neo4j:neo4j-connector-apache-spark_<scala>:*_for_spark_3 for Spark 3.x
• Scala version in artifact matches cluster runtime (2.13 only on 6.x)
• Credentials in Databricks secrets or env vars — not hardcoded
• node.keys set when using Overwrite mode
• Uniqueness constraint created on node.keys properties before MERGE writes
• coalesce(1) applied before relationship writes
• batch.size sized to Neo4j heap (start 5000, tune up)
• Delta Lake → Neo4j: nodes written before relationships
• query mode: no SKIP/LIMIT in Cypher (connector paginates internally)
• Databricks: Single User access mode (not Shared)