Neo4j Spark Skill

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

Maven artifact (Scala 2.12, Spark 3):

org.neo4j:neo4j-connector-apache-spark_2.12:5.4.2_for_spark_3

Scala 2.13 variant:

org.neo4j:neo4j-connector-apache-spark_2.13:5.4.2_for_spark_3

---

Setup

Standalone Spark (PySpark)

from pyspark.sql import SparkSession

spark = (SparkSession.builder
    .appName("neo4j-app")
    .config("spark.jars.packages",
            "org.neo4j:neo4j-connector-apache-spark_2.12:5.4.2_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:neo4j-connector-apache-spark_2.12:5.4.2_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()

Databricks — Cluster Installation

1. Cluster → Libraries → Install New → Maven 2. Search: org.neo4j:neo4j-connector-apache-spark_2.12 — match Scala version to runtime 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

---

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

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

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

# 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

---

Checklist

• [ ] Connector JAR version matches Spark version suffix (_for_spark_3)
• [ ] Scala version in artifact matches cluster runtime (2.12 vs 2.13)
• [ ] 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)

neo4j-spark-skill

Skill for reading and writing Neo4j data using the Neo4j Connector for Apache Spark, including Databricks, EMR, and standalone Spark environments.

Covers:

• SparkSession setup with Maven artifact org.neo4j:neo4j-connector-apache-spark
• DataFrame reads: label scan, Cypher query, relationship scan
• DataFrame writes: node CREATE/MERGE, relationship write with source/target mapping
• node.keys for Overwrite (MERGE) mode
• Partition and batch tuning (partitions, batch.size, schema.flatten.limit)
• Databricks cluster installation, secrets management, Unity Catalog notes
• Delta Lake → Neo4j ingestion pipeline pattern
• PySpark and Scala code examples

Version / Compatibility:

• Connector: 5.4.2_for_spark_3 (Scala 2.12 or 2.13)
• Spark: 3.3, 3.4, 3.5
• Databricks Runtime: 12.2, 13.3, 14.3 LTS
• Neo4j: 4.4, 5.x, 2025.x

Not covered:

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

Install:

npx skills add https://github.com/neo4j-contrib/neo4j-skills --skill neo4j-spark-skill

Or paste this link into your coding assistant: https://github.com/neo4j-contrib/neo4j-skills/tree/main/neo4j-spark-skill

Neo4j Spark Connector — Read Options Reference

Full option reference for .read.format("org.neo4j.spark.DataSource").

Core Read Options (mutually exclusive — pick one)

Label Read Sub-Options

Relationship Read Sub-Options

Query Read Sub-Options

Partition and Performance Options

Output Columns

Label scan result columns:

• <id> — internal Neo4j element ID
• <labels> — array of node labels
• One column per node property

Relationship scan result columns:

• <rel.id> — internal relationship ID
• <rel.type> — relationship type string
• <source.id>, <source.labels>, source.<prop> — source node fields
• <target.id>, <target.labels>, target.<prop> — target node fields
• Relationship property columns at top level

Schema Inference Notes

• Without APOC: samples schema.flatten.limit rows to infer types
• With APOC: uses apoc.meta.nodeTypeProperties — more accurate
• Map/list properties: flattened into dot-notation columns (e.g. address.city)
• Use query mode with explicit RETURN types when inference is unreliable

Examples

Multi-label AND filter

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

Cypher with explicit column types

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

Partitioned read for large node set

df = (spark.read.format("org.neo4j.spark.DataSource")
    .option("labels", ":Transaction")
    .option("partitions", "20")
    .option("batch.size", "10000")
    .option("query.count", "MATCH (n:Transaction) RETURN count(n)")
    .load())

Relationship with properties

df = (spark.read.format("org.neo4j.spark.DataSource")
    .option("relationship", "ACTED_IN")
    .option("relationship.source.labels", ":Person")
    .option("relationship.target.labels", ":Movie")
    .load())
# Columns: <rel.id>, <rel.type>, <source.id>, source.name, <target.id>, target.title, roles

Neo4j Spark Connector — Write Options Reference

Full option reference for .write.format("org.neo4j.spark.DataSource").

Save Modes

Core Write Options (mutually exclusive — pick one)

Node Write Options

Relationship Write Options

Node Keys Mapping Syntax

node.keys = "df_column"               # same name in graph property
node.keys = "df_column:graph_prop"    # rename
node.keys = "id,email"                # multiple keys (AND match in MERGE)
node.keys = "user_id:id,email:email"  # multiple with rename

Property Column Mapping Syntax

Same syntax for node.properties, relationship.properties, relationship.source.node.properties, relationship.target.node.properties:

"col1,col2"              # include these columns, use same names
"df_col:graph_prop"      # rename on write
"name,email:emailAddr"   # mix

Query Write Mode

DataFrame row values available via event.column_name:

write_query = """
    MERGE (p:Person {email: event.email})
    SET p.name = event.name, p.updatedAt = timestamp()
"""
(df.write.format("org.neo4j.spark.DataSource")
    .option("query", write_query)
    .mode("Overwrite")
    .save())

Performance Options

Relationship Node Save Modes

Full Relationship Write Example (Scala)

import org.apache.spark.sql.SaveMode

relDF.coalesce(1)
  .write
  .format("org.neo4j.spark.DataSource")
  .mode(SaveMode.Append)
  .option("relationship", "BOUGHT")
  .option("relationship.save.strategy", "keys")
  .option("relationship.source.labels", ":Customer")
  .option("relationship.source.save.mode", "Match")
  .option("relationship.source.node.keys", "cust_id:customerId")
  .option("relationship.target.labels", ":Product")
  .option("relationship.target.save.mode", "Match")
  .option("relationship.target.node.keys", "prod_id:productId")
  .option("relationship.properties", "qty:quantity,ts:purchasedAt")
  .save()

Pre-Write Checklist

• [ ] Uniqueness constraint on all node.keys / *.node.keys properties
• [ ] coalesce(1) before relationship write
• [ ] node.properties limits payload to needed columns
• [ ] batch.size validated against Neo4j heap
• [ ] Overwrite on nodes: constraint prevents duplicates under concurrency