neo4j-kafka-skill

Neo4j Kafka 技能

适用场景

• 将 Kafka 事件写入 Neo4j（Sink Connector — Cypher、Pattern、CDC、CUD 策略）
• 将 Neo4j 变更流式传输到 Kafka 主题（Source Connector — 基于 CDC 或查询）
• 无需 Kafka，直接查询 Neo4j 变更事件（db.cdc.query）
• 配置 Confluent Cloud 托管的 Neo4j Sink Connector
• 设置 Schema Registry（Avro/JSON Schema），用于处理类型化 Kafka 消息
• 在 Sink 上启用精确一次语义或死信队列

不适用场景

• 编写 Cypher 查询 → neo4j-cypher-skill
• 批量导入 CSV/JSON 文件 → neo4j-import-skill
• GDS 算法 → neo4j-gds-skill
• 实时应用写入模式 → neo4j-cypher-skill

---

决策表 — 应选择哪种 Connector 策略？

使用场景	策略
将 Kafka 负载自定义转换为图数据	Sink：Cypher
镜像另一个 Neo4j CDC 源	Sink：CDC（schema 或 source-id 子策略）
无需编写代码，将 Kafka JSON 字段映射到图节点/关系	Sink：Pattern
消费预先格式化的 CUD JSON 消息	Sink：CUD
将所有 Neo4j 变更实时流式传输到 Kafka	Source：CDC（Neo4j 5.13+ EE/Aura BC/VDC）
按计划流式传输特定查询结果	Source：Query
在进程内消费 CDC 事件，无需 Kafka	原生 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 Source/Sink：Neo4j 5.13+ Enterprise Edition、AuraDB Business Critical 或 AuraDB VDC
• 对于 Query Source：任意 Neo4j 版本
• Java 11+

---

核心连接配置（所有 Connector 通用）

{
  "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:...}）。

---

Sink Connector

策略 1 — Cypher

Connector 会自动添加前缀 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})"
}

Pattern 规则：

• !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"
}

精确一次语义（EOS）

要求：Connector ≥ 5.3.0（Cypher/Pattern/CDC 策略）、Connector ≥ 5.3.1（CUD 策略）、Kafka Broker 支持 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 — 添加到 Connector 配置：

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

不启用 EOS 时：Connector 提供至少一次语义 — 编写幂等 Cypher（使用 MERGE，而非 CREATE）。

错误处理 / 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。

---

Source Connector

基于 CDC 的 Source（推荐，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 | 特定的游标字符串

每个主题可配置多个模式 — 索引为 0、1、2...：

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

游标警告：从备份恢复数据库后，CDC 游标会失效。请重新配置 neo4j.start-from。

基于查询的 Source（旧版 / 适用于任意版本）

{
  "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"
}

Connector 会自动注入 $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": "..."
}

每个主题只能使用一种策略 — 同一主题上不能混用 Cypher 和 Pattern。

---

Schema Registry（Avro / JSON Schema）

源连接器始终生成带有架构支持的消息 — 必须配置转换器：

{
  "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 转换器必须与源端匹配 — Avro Sink 无法使用 JSON schema 源消息。

---

常见错误

错误	原因	修复方法
CDC is not enabled	未设置 db.cdc.enabled / 层级不正确	在 neo4j.conf 中启用，或升级到 EE/BC/VDC
数据库恢复后出现 Invalid cursor	备份会使游标失效	将 neo4j.start-from 重置为 NOW 或 EARLIEST
Cannot merge node using null	消息中缺少键属性	验证消息架构；在 Cypher 中添加空值检查
重启后重放消息	未配置 EOS	添加 neo4j.eos-offset-label + NODE KEY 约束
连接器因错误消息而停止	errors.tolerance=none（默认值）	设置 errors.tolerance=all + DLQ 主题
Sink 上出现 SchemaException	源端和 Sink 端转换器不匹配	匹配两端的键/值转换器
db.cdc.query 返回空事件	游标指向当前位置	使用 db.cdc.earliest() 重放；等待新事务

---

参考资料

• 完整连接器配置参考 — 所有 neo4j.* 属性、默认值和类型
• CDC API 模式 — 游标循环、选择器示例、事件结构详情
• Neo4j Connector for Kafka 文档
• CDC 文档

---

检查清单

• 如果使用 CDC 源或 Sink，已确认 CDC 可用（Neo4j 5.13+ EE / Aura BC / VDC）
• 在导入到 Sink 之前已创建唯一性约束/NODE KEY 约束（MERGE 会使用这些约束）
• 如果使用 neo4j.eos-offset-label，已创建 EOS 约束
• 通过密钥提供程序提供凭据 — 不要在配置中硬编码
• Cypher Sink 查询使用 MERGE（而非 CREATE）以确保幂等性
• 已为生产环境的 Sink 配置 errors.tolerance=all + DLQ
• 源：neo4j.query.streaming-property 已建立索引
• 源端和 Sink 端的 Schema Registry 转换器相匹配
• 数据库恢复后：已重新配置 CDC 游标（neo4j.start-from）
• CDC 游标循环：仅在处理成功后推进游标