neo4j-spark-skill

Apache Spark 用 Neo4j コネクター

使用する場面

• Neo4j のノード／リレーションシップを Spark DataFrame に読み込む
• Spark DataFrame をノードまたはリレーションシップとして Neo4j に書き込む
• Neo4j に接続する Databricks ノートブック
• 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 を参照するリロケーション 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（+s／+ssc URI では無視されます）	false
neo4j.db.transaction.timeout	トランザクションのタイムアウト（ms）	ドライバーのデフォルト
neo4j.db.transaction.metadata.<key>	クエリログに表示されるカスタムトランザクションメタデータ［6.0］	空
neo4j.authentication.type = supplier 名	有効期限付き OAuth／OIDC トークン用のカスタム AuthenticationTokenSupplierFactory（例：org.neo4j.connectors:commons-authn-keycloak による keycloak）	—

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 からの読み取り

相互に排他的な読み取りモードが 3 つあります。.read() 呼び出しごとに、必ず 1 つだけ使用してください。

ラベルスキャン（ノード）

# 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 を 1 回実行します。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 は併用できません。1 つの script 内でセミコロン区切りのステートメントを指定すると、6.0 では失敗します。

書き込みオプションのリファレンス全文：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 に減らし、ヒープを確認します
スキーマの列がすべて string になる	APOC がなく、スキーマがサンプリングされていない	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 のデフォルト戦略が native ではなく keys	.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 バージョンがクラスターのランタイムと一致している（2.13 は 6.x でのみ使用）
• 認証情報は Databricks のシークレットまたは環境変数に設定し、ハードコードしない
• Overwrite モードを使用する場合は node.keys を設定する
• MERGE 書き込みの前に、node.keys のプロパティに一意性制約を作成する
• リレーションシップを書き込む前に coalesce(1) を適用する
• batch.size は Neo4j のヒープに合わせて設定する（5000 から始めて調整する）
• Delta Lake → Neo4j：リレーションシップより先にノードを書き込む
• query モード：Cypher に SKIP／LIMIT を指定しない（コネクターが内部でページ分割する）
• Databricks：Single User アクセスモードを使用する（Shared ではない）