neo4j-kafka-skill

Neo4j Kafka スキル

使用する場面

• Kafka イベントを Neo4j に書き込む場合（シンクコネクター — Cypher、Pattern、CDC、CUD 戦略）
• Neo4j の変更を Kafka トピックにストリーミングする場合（ソースコネクター — CDC またはクエリベース）
• Kafka を使わずに Neo4j の変更イベントをネイティブにクエリする場合（db.cdc.query）
• Confluent Cloud のマネージド Neo4j シンクコネクターを設定する場合
• 型付き Kafka メッセージ向けにスキーマレジストリ（Avro／JSON Schema）を設定する場合
• シンクで exactly-once セマンティクスまたはデッドレターキューを有効にする場合

使用しない場面

• Cypher クエリの作成 → neo4j-cypher-skill
• CSV／JSON ファイルの一括インポート → neo4j-import-skill
• GDS アルゴリズム → neo4j-gds-skill
• ライブアプリの書き込みパターン → neo4j-cypher-skill

---

選択表 — どのコネクター戦略を使うか？

ユースケース	戦略
Kafka ペイロードをグラフにカスタム変換する	シンク：Cypher
別の Neo4j CDC ソースをミラーリングする	シンク：CDC（schema または source-id サブ戦略）
コードを書かずに Kafka JSON フィールドをグラフのノード／リレーションにマッピングする	シンク：Pattern
事前フォーマット済みの CUD JSON メッセージを消費する	シンク：CUD
すべての Neo4j の変更を Kafka にストリーミングする（リアルタイム）	ソース：CDC（Neo4j 5.13+ EE／Aura BC／VDC）
特定のクエリ結果をスケジュールに従ってストリーミングする	ソース：Query
Kafka を使わず、プロセス内で CDC イベントを消費する	ネイティブ CDC API（db.cdc.query）

---

前提条件

• Neo4j Connector for Kafka ≥ 5.5.2（neo4j.com/labs/kafka または Confluent Hub からダウンロード）— 5.5.2 はトピック単位ではなくパーティション単位で受信メッセージを処理します。5.5.1 では生成されるクエリに静的ラベルが強制されます
• Kafka Connect ≥ 3.x または Confluent Platform ≥ 7.x
• CDC ソース／シンクの場合：Neo4j 5.13+ Enterprise Edition、AuraDB Business Critical、または AuraDB VDC
• クエリソースの場合：すべての Neo4j エディション
• Java 11+

---

コア接続設定（すべてのコネクター）

{
  "neo4j.uri": "neo4j+s://your-instance.databases.neo4j.io:7687",
  "neo4j.authentication.type": "BASIC",
  "neo4j.authentication.basic.username": "neo4j",
  "neo4j.authentication.basic.password": "${file:/opt/secrets.properties:neo4j.password}",
  "neo4j.database": "neo4j"
}

認証タイプ：BASIC | BEARER | KERBEROS | CUSTOM | NONE

パスワードをハードコードしないでください — Kafka Connect のシークレットプロバイダー（${file:...} または ${env:...}）を使用してください。

---

シンクコネクター

戦略 1 — Cypher

コネクターは自動的に UNWIND $events AS __value を先頭に追加します — __value を使って書き込みクエリを作成してください。

{
  "connector.class": "org.neo4j.connectors.kafka.sink.Neo4jConnector",
  "topics": "person-creates,person-updates",
  "neo4j.uri": "neo4j+s://...",
  "neo4j.authentication.type": "BASIC",
  "neo4j.authentication.basic.username": "neo4j",
  "neo4j.authentication.basic.password": "secret",
  "neo4j.cypher.topic.person-creates":
    "MERGE (p:Person {id: __value.id}) SET p += __value.properties",
  "neo4j.cypher.topic.person-updates":
    "MATCH (p:Person {id: __value.id}) SET p += __value.properties",
  "neo4j.cypher.bind-value-as": "__value",
  "neo4j.cypher.bind-key-as": "__key",
  "neo4j.cypher.bind-header-as": "__header"
}

MERGE パターン — 冪等なアップサート：

MERGE (p:Person {id: __value.id})
ON CREATE SET p.createdAt = datetime(), p += __value.properties
ON MATCH  SET p.updatedAt = datetime(), p += __value.properties

戦略 2 — Pattern

Cypher は不要です — パターン構文でメッセージフィールドをグラフにマッピングします。

{
  "neo4j.pattern.topic.users": "(:User{!userId, name, email})",
  "neo4j.pattern.topic.friendships":
    "(:User{!userId: from.userId})-[:KNOWS{since}]->(:User{!userId: to.userId})"
}

パターンのルール：

• !prop = キープロパティ（MERGE に使用）
• prop: field.path = ネストされたメッセージフィールドからマッピング
• * = すべてのメッセージフィールドをマッピング
• -prop = プロパティを除外（包含指定との併用不可）

戦略 3 — CDC（別の Neo4j をミラーリング）

{
  "neo4j.cdc.schema.topics": "neo4j-cdc-events"
}

または source-id トラッキングを使用します（elementId をプロパティとして保存）：

{
  "neo4j.cdc.source-id.topics": "neo4j-cdc-events",
  "neo4j.cdc.source-id.label-name": "SourceEvent",
  "neo4j.cdc.source-id.property-name": "sourceId"
}

Exactly-Once セマンティクス（EOS）

必要条件：コネクター ≥ 5.3.0（Cypher／Pattern／CDC 戦略）、コネクター ≥ 5.3.1（CUD 戦略）、Kafka ブローカーの EOS サポート、および NODE KEY 制約。

ステップ 1 — 制約を作成します：

CREATE CONSTRAINT kafka_offset_key IF NOT EXISTS
FOR (n:__KafkaOffset)
REQUIRE (n.strategy, n.topic, n.partition) IS NODE KEY;

ステップ 2 — コネクター設定に追加します：

{
  "neo4j.eos-offset-label": "__KafkaOffset"
}

EOS を使用しない場合：コネクターは at-least-once を提供します — 冪等な Cypher（CREATE ではなく MERGE）を書いてください。

エラー処理／DLQ

{
  "errors.tolerance": "all",
  "errors.log.enable": "true",
  "errors.log.include.messages": "true",
  "errors.deadletterqueue.topic.name": "neo4j-dlq",
  "errors.deadletterqueue.context.headers.enable": "true",
  "errors.deadletterqueue.topic.replication.factor": "3"
}

errors.tolerance=none（デフォルト）— 最初のエラーで停止します。本番環境では all と DLQ を使用してください。

---

ソースコネクター

CDC ベースのソース（推奨、Neo4j 5.13+）

{
  "connector.class": "org.neo4j.connectors.kafka.source.Neo4jConnector",
  "neo4j.uri": "neo4j+s://...",
  "neo4j.authentication.type": "BASIC",
  "neo4j.authentication.basic.username": "neo4j",
  "neo4j.authentication.basic.password": "secret",
  "neo4j.source-strategy": "CDC",
  "neo4j.start-from": "NOW",
  "neo4j.cdc.poll-interval": "1s",
  "neo4j.cdc.poll-duration": "5s",
  "neo4j.cdc.topic.person-creates.patterns.0.pattern": "(:Person)",
  "neo4j.cdc.topic.person-creates.patterns.0.operation": "CREATE",
  "neo4j.cdc.topic.person-updates.patterns.0.pattern": "(:Person)",
  "neo4j.cdc.topic.person-updates.patterns.0.operation": "UPDATE",
  "neo4j.cdc.topic.person-deletes.patterns.0.pattern": "(:Person)",
  "neo4j.cdc.topic.person-deletes.patterns.0.operation": "DELETE"
}

neo4j.start-from のオプション：NOW | EARLIEST | 特定のカーソル文字列

1 つのトピックに複数のパターンを指定できます — 0、1、2... のインデックスを付けます：

{
  "neo4j.cdc.topic.all-changes.patterns.0.pattern": "(:Person)",
  "neo4j.cdc.topic.all-changes.patterns.1.pattern": "(:Organization)"
}

カーソルに関する警告：バックアップから DB を復元すると、CDC カーソルは無効になります。neo4j.start-from を再設定してください。

クエリベースのソース（レガシー／すべてのエディション）

{
  "neo4j.source-strategy": "QUERY",
  "neo4j.query": "MATCH (p:Person) WHERE p.updatedAt > $lastCheck RETURN p.id AS id, p.name AS name, p.updatedAt AS updatedAt",
  "neo4j.query.streaming-property": "updatedAt",
  "neo4j.query.topic": "person-changes",
  "neo4j.query.polling-interval": "5s",
  "neo4j.query.polling-duration": "10s"
}

$lastCheck はコネクターによって自動的に注入されます。neo4j.query.streaming-property はクエリの戻り値に含める必要があり、インデックスを付けることを推奨します。

---

ネイティブ CDC API（Kafka 不要）

必要条件：Neo4j 5.13+ Enterprise、AuraDB BC、または AuraDB VDC。

まず CDC を有効にします（セルフマネージド — neo4j.conf に設定）：

db.cdc.enabled=true

Aura では、対象ティアでデフォルトで有効になっています。

カーソルの初期化

// Get cursor for "right now" — start tracking from this point forward
CALL db.cdc.current() YIELD id RETURN id AS cursor;

// Get earliest available cursor (replay from history start)
CALL db.cdc.earliest() YIELD id RETURN id AS cursor;

カーソルは排他的です。db.cdc.current() には、カーソルが指すトランザクションは含まれません。

クエリの変更

// All changes since cursor
CALL db.cdc.query($cursor, []) YIELD id, txId, seq, metadata, event
RETURN id, txId, seq, metadata, event
ORDER BY txId, seq;

フィルター適用済み — ラベルが Person のノードのみ、CREATE のみ：

CALL db.cdc.query($cursor, [
  {select: 'n', labels: ['Person'], operation: 'c'}
]) YIELD id, txId, seq, event
RETURN id, event.state.after.properties AS newProps
ORDER BY txId, seq;

フィルター適用済み — プロパティ変更の追跡を伴う、特定のリレーションシップタイプ：

CALL db.cdc.query($cursor, [
  {select: 'r', type: 'KNOWS', changesTo: ['since', 'strength']}
]) YIELD id, txId, seq, event
RETURN id, event.state.before AS before, event.state.after AS after;

セレクターのリファレンス

フィールド	値	適用対象
select	'e'（すべて）、'n'（ノード）、'r'（リレーションシップ）	両方
operation	'c'（作成）、'u'（更新）、'd'（削除）	両方
labels	['Label1','Label2']（ノードはすべてを持つ必要があります）	ノード
type	'REL_TYPE'	リレーションシップ
elementId	特定の要素 ID 文字列	両方
key	{propName: value}（キー制約が必要です）	両方
changesTo	['prop1','prop2']（AND — すべてが変更される必要があります）	両方
authenticatedUser	ユーザー名文字列	両方
executingUser	ユーザー名文字列	両方
txMetadata	{key: value}	両方

イベント構造

{
  id:       STRING,           // cursor for this event (use as next $cursor)
  txId:     INTEGER,          // transaction ID
  seq:      INTEGER,          // ordering within transaction
  metadata: {
    executingUser:     STRING,
    authenticatedUser: STRING,
    captureMode:       STRING,  // "DIFF" or "FULL"
    txStartTime:       DATETIME,
    txCommitTime:      DATETIME,
    txMetadata:        MAP
  },
  event: {
    elementId:  STRING,
    eventType:  STRING,         // "n" or "r"
    operation:  STRING,         // "c", "u", "d"
    labels:     [STRING],       // nodes only
    type:       STRING,         // relationships only
    keys:       MAP,
    state: {
      before: { properties: MAP },  // null on CREATE
      after:  { properties: MAP }   // null on DELETE
    }
  }
}

カーソルループパターン（Python）

from neo4j import GraphDatabase

driver = GraphDatabase.driver("neo4j+s://...", auth=("neo4j", "password"))

def poll_changes(cursor: str, selectors: list) -> tuple[list, str]:
    records, _, _ = driver.execute_query(
        "CALL db.cdc.query($cursor, $selectors) YIELD id, txId, seq, event "
        "RETURN id, txId, seq, event ORDER BY txId, seq",
        cursor=cursor, selectors=selectors,
        database_="neo4j"
    )
    events = [r.data() for r in records]
    # Advance cursor to last event id; keep current if no events
    next_cursor = events[-1]["id"] if events else cursor
    return events, next_cursor

# Bootstrap
with driver.session(database="neo4j") as s:
    cursor = s.run("CALL db.cdc.current() YIELD id RETURN id").single()["id"]

selectors = [{"select": "n", "labels": ["Person"]}]

import time
while True:
    events, cursor = poll_changes(cursor, selectors)
    for e in events:
        print(e["event"]["operation"], e["event"]["elementId"])
    time.sleep(1)

---

Confluent Cloud マネージドコネクター

Confluent Cloud は、Neo4j Sink コネクターをフルマネージドサービスとしてホストしています（JAR のアップロードは不要です）。

セルフマネージドとの設定の違い：

• connector.class フィールドはありません — UI/API で選択します
• 認証情報は Confluent Cloud シークレットマネージャーまたは直接 JSON で指定します
• プライベートエンドポイントに対応しています（AWS PrivateLink、Azure Private Link、GCP PSC）
• マネージドアップグレード — 必要に応じてコネクターのバージョンを明示的に固定します

Confluent Cloud で必須のフィールド：

{
  "kafka.auth.mode": "KAFKA_API_KEY",
  "kafka.api.key": "...",
  "kafka.api.secret": "...",
  "input.data.format": "JSON",
  "neo4j.uri": "neo4j+s://...",
  "neo4j.authentication.type": "BASIC",
  "neo4j.authentication.basic.username": "neo4j",
  "neo4j.authentication.basic.password": "..."
}

トピックごとに 1 つの戦略のみ使用できます。同じトピックで Cypher と Pattern を混在させることはできません。

---

Schema Registry（Avro / JSON Schema）

Source コネクターは常にスキーマ対応のメッセージを生成するため、コンバーターを設定する必要があります：

{
  "key.converter": "io.confluent.connect.avro.AvroConverter",
  "key.converter.schema.registry.url": "https://your-schema-registry",
  "value.converter": "io.confluent.connect.avro.AvroConverter",
  "value.converter.schema.registry.url": "https://your-schema-registry"
}

JSON Schema の場合：

{
  "value.converter": "io.confluent.connect.json.JsonSchemaConverter",
  "value.converter.schema.registry.url": "https://..."
}

Sink コンバーターは Source と一致させる必要があります。Avro Sink は JSON Schema の Source メッセージを消費できません。

---

よくあるエラー

エラー	原因	修正方法
CDC is not enabled	db.cdc.enabled が設定されていない／誤ったティア	neo4j.conf で有効にするか、EE/BC/VDC にアップグレードします
DB の復元後に発生する Invalid cursor	バックアップによってカーソルが無効になります	neo4j.start-from を NOW または EARLIEST にリセットします
Cannot merge node using null	メッセージにキーのプロパティがありません	メッセージスキーマを検証し、Cypher に null チェックを追加します
再起動後にメッセージが再生される	EOS が設定されていません	neo4j.eos-offset-label と NODE KEY 制約を追加します
不正なメッセージでコネクターが停止する	errors.tolerance=none（デフォルト）	errors.tolerance=all と DLQ トピックを設定します
Sink で SchemaException が発生する	Source/Sink 間でコンバーターが一致していません	両端でキー／値のコンバーターを一致させます
db.cdc.query から空のイベントが返される	カーソルが現在を指しています	再生するには db.cdc.earliest() を使用します。新しいトランザクションが発生するまで待ちます

---

参考資料

• コネクター設定の完全なリファレンス — すべての neo4j.* プロパティ、デフォルト値、型
• CDC API のパターン — カーソルループ、セレクターの例、イベント構造の詳細
• Neo4j Connector for Kafka のドキュメント
• CDC のドキュメント

---

チェックリスト

• CDC を使用する Source または Sink の場合、CDC の利用可否を確認済み（Neo4j 5.13+ EE / Aura BC / VDC）
• Sink のインポート前に一意性／NODE KEY 制約を作成済み（MERGE で使用）
• neo4j.eos-offset-label を使用する場合、EOS 制約を作成済み
• 認証情報はシークレットプロバイダー経由で指定 — 設定にハードコードしない
• Cypher Sink クエリでは、冪等性のために CREATE ではなく MERGE を使用
• 本番環境の Sink に errors.tolerance=all と DLQ を設定済み
• Source：neo4j.query.streaming-property にインデックスを設定済み
• Schema Registry コンバーターは Source と Sink の両側で一致
• DB の復元後：CDC カーソルを再設定（neo4j.start-from）
• CDC カーソルループ：処理が成功した後にのみカーソルを進める