neo4j-spark-skill

Apache Spark 的 Neo4j 连接器

适用场景

• 将 Neo4j 节点/关系读入 Spark DataFrame
• 将 Spark DataFrame 写入 Neo4j，作为节点或关系
• Databricks 笔记本连接 Neo4j
• Delta Lake → Neo4j 数据摄取管道
• 从大型 Neo4j 图中分区并行读取

不适用场景

• Python bolt 驱动程序 / execute_query → neo4j-driver-python-skill
• Cypher 查询编写 → neo4j-cypher-skill
• GDS 图算法 → neo4j-gds-skill
• Spring Boot + Neo4j → neo4j-spring-data-skill

---

版本矩阵

连接器	Spark	Scala	Java	Databricks Runtime	Neo4j	Maven 坐标
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

6.0 中更改了 Group ID——org.neo4j:neo4j-connector-apache-spark_<scala> 现在是指向 org.neo4j.connectors:spark 的 relocation POM。使用 Spark 3.x 时，请继续使用 5.5.x。

6.0 重大变更

变更	迁移方式
Spark 基准版本从 3.5 升至 4.0/4.1；不再支持 Scala 2.12 和 Java 8–11	先升级到 5.5.0，然后升级到 Spark 4.x + Scala 2.13 + Java 17
Maven 坐标为 org.neo4j.connectors:spark:<version>-s_2.13	替换旧的 _for_spark_3 坐标
移除 schema.optimization.type	使用 schema.optimization.node.keys、schema.optimization.relationship.keys、schema.optimization
分区读取中移除 $stream.offset	使用 partitions + query.count
移除以 ; 分隔的多语句 script	使用 script.1、script.2、…、script.N——按编号顺序执行
relationship.save.strategy 的默认值从 native 改为 keys	若要保留旧行为，请显式设置 .option("relationship.save.strategy", "native")
为 Data Source V2 谓词下推重写了 query 选项	无需操作；升级时请检查执行计划

---

设置

独立 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())

独立 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 — 集群安装

1. 集群 → Libraries → Install New → Maven
2. 在 DBR 17.3 LTS 上使用坐标 org.neo4j.connectors:spark:6.0.0-s_2.13；在 DBR 14.3–16.4 LTS 上使用 org.neo4j:neo4j-connector-apache-spark_2.13:5.5.0_for_spark_3
3. 集群 → Advanced Options → Spark 选项卡 — 添加配置：

   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. 使用 Single user 访问模式（不支持 Unity Catalog 共享模式）

Databricks — 密钥（优先于明文）

# 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)

---

关键配置选项

选项	说明	默认值
neo4j.url	Bolt/Neo4j URI	—（必填）
neo4j.authentication.type	none、basic、kerberos、bearer	basic
neo4j.authentication.basic.username	用户名	驱动程序默认值
neo4j.authentication.basic.password	密码	驱动程序默认值
neo4j.authentication.bearer.token	Bearer 令牌	—
neo4j.database	目标数据库	驱动程序默认值
neo4j.access.mode	read 或 write	read
neo4j.encryption.enabled	TLS（URI 使用 +s/+ssc 时忽略此项）	false
neo4j.db.transaction.timeout	事务超时时间（ms）	驱动程序默认值
neo4j.db.transaction.metadata.<key>	自定义事务元数据，可在查询日志中查看 [6.0]	空
neo4j.authentication.type = 供应器名称	自定义 AuthenticationTokenSupplierFactory（例如通过 org.neo4j.connectors:commons-authn-keycloak 使用 keycloak），用于会过期的 OAuth/OIDC 令牌	—

Cypher 版本和查询调优 [6.0]

选项	效果
cypher.version	Cypher 语言版本——5（默认）或 25
cypher.tuning.<param>	在每条生成的查询前添加 CYPHER <param>=<value> 前缀

读取和写入时可与 labels、relationship、query 一起使用；与 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())

---

从 Neo4j 读取

三种互斥的读取模式——每次 .read() 调用只能使用其中一种。

标签扫描（节点）

# 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()

多标签筛选（AND）：.option("labels", ":Person:Employee")

结果包含 <id>（Neo4j 内部 ID）和 <labels> 列。

Cypher 查询读取

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())

使用显式的 RETURN 别名——它们将成为 DataFrame 列名。查询中不能使用 SKIP/LIMIT（连接器会处理分页）。

关系扫描

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

结果列：<rel.id>、<rel.type>、<source.*>、<target.*>，以及关系属性。

读取分区调优

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())

完整读取选项参考：references/read-patterns.md

---

写入 Neo4j

SaveMode

SaveMode	Cypher	要求
Append	CREATE	无额外要求
Overwrite	MERGE	node.keys（节点）或 *.node.keys（关系）
ErrorIfExists	CREATE + 存在时出错	—

在 Overwrite 模式下写入前，始终先为 node.keys 属性创建唯一性约束。

写入节点 — 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())

写入节点 — 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：.option("node.keys", "df_col:node_property,id:personId")

写入节点 — 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()

写入关系

写入关系前使用 coalesce(1)，以避免死锁。

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 — 查找现有节点（缺失时失败）
• Append — 始终 CREATE 新节点
• Overwrite — MERGE 节点

写入前脚本 [6.0]

script.N 会在写入操作前按编号顺序运行一次 Cypher。使用 query 模式时，索引/约束设置需要此项（该模式不接受 schema.optimization.*）。

(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（单条语句）和 script.N 不能同时使用。在 6.0 中，一个 script 内以分号分隔多条语句会失败。

完整写入选项参考：references/write-patterns.md

---

Databricks — Delta Lake → Neo4j 流水线

# 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())

关系的流水线模式——先加载两组节点，再写入边：

# 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()

---

写入性能调优

场景	建议
节点写入（无锁竞争）	repartition(N)，其中 N ≤ Neo4j CPU 核数
关系写入（有锁风险）	coalesce(1) — 单个分区
大型数据集	batch.size 设为 10000–20000（根据堆内存调整）
大量 MERGE 的加载	先为 node.keys 属性添加唯一性约束

# 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())

---

常见错误

错误	原因	解决方法
ClassNotFoundException: org.neo4j.spark.DataSource	JAR 不在类路径中	添加 spark.jars.packages 或附加库
写入关系时发生死锁	多个分区锁定节点	写入前使用 coalesce(1)
Overwrite 时出现重复节点	键没有唯一性约束	CREATE CONSTRAINT ON (n:Label) ASSERT n.prop IS UNIQUE
Neo4j 端 OOM	batch.size 过大	降至 5000–10000；检查堆内存
Schema 中所有列都是 string	没有 APOC，未采样 Schema	提高 schema.flatten.limit；或使用带显式类型的 query 模式
写入时出现 Access mode is read 错误	会话以读取模式打开	移除 neo4j.access.mode 或设为 write
Databricks Shared 集群失败	不支持 Unity Catalog 共享模式	切换到 Single User 访问模式
Spark 4 上出现 NoSuchMethodError / IncompatibleClassChangeError	Spark 4 运行时使用了 5.x 连接器	使用 org.neo4j.connectors:spark:6.0.0-s_2.13
升级后关系写入忽略 rel.* / source.* 列	6.0 的默认策略是 keys，而不是 native	.option("relationship.save.strategy", "native")
有多条语句时 script 选项被拒绝	6.0 移除了以 ; 分隔的脚本	拆分为 script.1、script.2、…

---

检查清单

• 连接器坐标与 Spark 版本匹配——Spark 4.x 使用 org.neo4j.connectors:spark:*-s_2.13，Spark 3.x 使用 org.neo4j:neo4j-connector-apache-spark_<scala>:*_for_spark_3
• 工件中的 Scala 版本与集群运行时匹配（仅在 6.x 上使用 2.13）
• 凭据存储在 Databricks secrets 或环境变量中——不要硬编码
• 使用 Overwrite 模式时设置 node.keys
• 执行 MERGE 写入前，在 node.keys 属性上创建唯一性约束
• 写入关系前应用 coalesce(1)
• 根据 Neo4j 堆大小设置 batch.size（从 5000 开始，再调整）
• Delta Lake → Neo4j：先写入节点，再写入关系
• query 模式：Cypher 中不要包含 SKIP/LIMIT（连接器会在内部进行分页）
• Databricks：使用 Single User 访问模式（而非 Shared）