neo4j-kafka-skill
installsforks
How to install neo4j-kafka-skill?
neo4j-contrib/neo4j-skills/neo4j-kafka-skill
$npx -y skills add neo4j-contrib/neo4j-skills --skill neo4j-kafka-skillConfigure and operate the Neo4j Connector for Kafka (sink + source) and the
SKILL.md
| 1 | # Neo4j Kafka Skill |
| 2 | |
| 3 | ## When to Use |
| 4 | |
| 5 | - Writing Kafka events into Neo4j (sink connector — Cypher, Pattern, CDC, CUD strategies) |
| 6 | - Streaming Neo4j changes to Kafka topics (source connector — CDC or query-based) |
| 7 | - Querying Neo4j change events natively without Kafka (`db.cdc.query`) |
| 8 | - Configuring Confluent Cloud managed Neo4j sink connector |
| 9 | - Setting up schema registry (Avro/JSON Schema) for typed Kafka messages |
| 10 | - Enabling exactly-once semantics or dead-letter queue on sink |
| 11 | |
| 12 | ## When NOT to Use |
| 13 | |
| 14 | - **Cypher query authoring** → `neo4j-cypher-skill` |
| 15 | - **Bulk CSV/JSON file import** → `neo4j-import-skill` |
| 16 | - **GDS algorithms** → `neo4j-gds-skill` |
| 17 | - **Live app write patterns** → `neo4j-cypher-skill` |
| 18 | |
| 19 | --- |
| 20 | |
| 21 | ## Decision Table — Which connector strategy? |
| 22 | |
| 23 | | Use case | Strategy | |
| 24 | |---|---| |
| 25 | | Custom transformation of Kafka payload → graph | Sink: **Cypher** | |
| 26 | | Mirror another Neo4j CDC source | Sink: **CDC** (schema or source-id sub-strategy) | |
| 27 | | Map Kafka JSON fields to graph nodes/rels with no code | Sink: **Pattern** | |
| 28 | | Consume pre-formatted CUD JSON messages | Sink: **CUD** | |
| 29 | | Stream all Neo4j changes to Kafka (real-time) | Source: **CDC** (Neo4j 5.13+ EE/Aura BC/VDC) | |
| 30 | | Stream specific query results on a schedule | Source: **Query** | |
| 31 | | Consume CDC events in-process, no Kafka | **Native CDC API** (`db.cdc.query`) | |
| 32 | |
| 33 | --- |
| 34 | |
| 35 | ## Prerequisites |
| 36 | |
| 37 | - Neo4j Connector for Kafka ≥ 5.0 (download from [neo4j.com/labs/kafka](https://neo4j.com/labs/kafka/) or Confluent Hub) |
| 38 | - Kafka Connect ≥ 3.x or Confluent Platform ≥ 7.x |
| 39 | - For CDC source/sink: Neo4j 5.13+ Enterprise Edition, AuraDB Business Critical, or AuraDB VDC |
| 40 | - For query source: any Neo4j edition |
| 41 | - Java 11+ |
| 42 | |
| 43 | --- |
| 44 | |
| 45 | ## Core Connection Config (all connectors) |
| 46 | |
| 47 | ```json |
| 48 | { |
| 49 | "neo4j.uri": "neo4j+s://your-instance.databases.neo4j.io:7687", |
| 50 | "neo4j.authentication.type": "BASIC", |
| 51 | "neo4j.authentication.basic.username": "neo4j", |
| 52 | "neo4j.authentication.basic.password": "${file:/opt/secrets.properties:neo4j.password}", |
| 53 | "neo4j.database": "neo4j" |
| 54 | } |
| 55 | ``` |
| 56 | |
| 57 | Authentication types: `BASIC` | `BEARER` | `KERBEROS` | `CUSTOM` | `NONE` |
| 58 | |
| 59 | Never hardcode passwords — use Kafka Connect secrets provider (`${file:...}` or `${env:...}`). |
| 60 | |
| 61 | --- |
| 62 | |
| 63 | ## Sink Connector |
| 64 | |
| 65 | ### Strategy 1 — Cypher |
| 66 | |
| 67 | Connector auto-prepends `UNWIND $events AS __value` — write query using `__value`: |
| 68 | |
| 69 | ```json |
| 70 | { |
| 71 | "connector.class": "org.neo4j.connectors.kafka.sink.Neo4jConnector", |
| 72 | "topics": "person-creates,person-updates", |
| 73 | "neo4j.uri": "neo4j+s://...", |
| 74 | "neo4j.authentication.type": "BASIC", |
| 75 | "neo4j.authentication.basic.username": "neo4j", |
| 76 | "neo4j.authentication.basic.password": "secret", |
| 77 | "neo4j.cypher.topic.person-creates": |
| 78 | "MERGE (p:Person {id: __value.id}) SET p += __value.properties", |
| 79 | "neo4j.cypher.topic.person-updates": |
| 80 | "MATCH (p:Person {id: __value.id}) SET p += __value.properties", |
| 81 | "neo4j.cypher.bind-value-as": "__value", |
| 82 | "neo4j.cypher.bind-key-as": "__key", |
| 83 | "neo4j.cypher.bind-header-as": "__header" |
| 84 | } |
| 85 | ``` |
| 86 | |
| 87 | MERGE pattern — idempotent upsert: |
| 88 | ```cypher |
| 89 | MERGE (p:Person {id: __value.id}) |
| 90 | ON CREATE SET p.createdAt = datetime(), p += __value.properties |
| 91 | ON MATCH SET p.updatedAt = datetime(), p += __value.properties |
| 92 | ``` |
| 93 | |
| 94 | ### Strategy 2 — Pattern |
| 95 | |
| 96 | No Cypher needed — map message fields to graph via pattern syntax: |
| 97 | |
| 98 | ```json |
| 99 | { |
| 100 | "neo4j.pattern.topic.users": "(:User{!userId, name, email})", |
| 101 | "neo4j.pattern.topic.friendships": |
| 102 | "(:User{!userId: from.userId})-[:KNOWS{since}]->(:User{!userId: to.userId})" |
| 103 | } |
| 104 | ``` |
| 105 | |
| 106 | Pattern rules: |
| 107 | - `!prop` = key property (used for MERGE) |
| 108 | - `prop: field.path` = map from nested message field |
| 109 | - `*` = map all message fields |
| 110 | - `-prop` = exclude property (cannot mix with inclusions) |
| 111 | |
| 112 | ### Strategy 3 — CDC (mirror another Neo4j) |
| 113 | |
| 114 | ```json |
| 115 | { |
| 116 | "neo4j.cdc.schema.topics": "neo4j-cdc-events" |
| 117 | } |
| 118 | ``` |
| 119 | |
| 120 | Or with source-id tracking (stores elementId as property): |
| 121 | ```json |
| 122 | { |
| 123 | "neo4j.cdc.source-id.topics": "neo4j-cdc-events", |
| 124 | "neo4j.cdc.source-id.label-name": "SourceEvent", |
| 125 | "neo4j.cdc.source-id.property-name": "sourceId" |
| 126 | } |
| 127 | ``` |
| 128 | |
| 129 | ### Exactly-Once Semantics (EOS) |
| 130 | |
| 131 | Requires: connector ≥ 5.3.0 (Cypher/Pattern/CDC strategies), connector ≥ 5.3.1 (CUD strategy), Kafka broker EOS support, and a NODE KEY constraint. |
| 132 | |
| 133 | Step 1 — Create constraint: |
| 134 | ```cyphe |
People who liked this also viewed
6 picksType
- azure-storageAzure Storage Services including Blob Storage, File Shares, Queue Storage, Table Storage, and Data Lake. Answers questions about storage access tiers (hot, cool, cold, archive), when to use each tier,SkillsJul 2026
- supabase-postgres-best-practicesPostgres performance optimization and best practices from Supabase. Use this skill when writing, reviewing, or optimizing Postgres queries, schema designs, or database configurations.SkillsJul 2026
- supabaseUse when doing ANY task involving Supabase. Triggers: Supabase products (Database, Auth, Edge Functions, Realtime, Storage, Vectors, Cron, Queues); client libraries and SSR integrations (supabase-js, SkillsJul 2026
- convex-quickstartCreates or adds Convex to an app. Use for new Convex projects, npm create convex@latest, frontend setup, env vars, or the first npx convex dev run.SkillsJul 2026
- convex-performance-auditAudits Convex performance for reads, subscriptions, write contention, and function limits. Use for slow features, insights findings, OCC conflicts, or read amplification.SkillsJul 2026
- convex-setup-authSets up Convex auth, identity mapping, and access control. Use for login, auth providers, users tables, protected functions, or roles in a Convex app.SkillsJul 2026
