Source: https://dev-docs.yueshu.com.cn/v5.3.2/clients/kafka-connector/
Documentation version: v5.3.2

# Kafka 连接器指南

[Kafka 连接器](https://github.com/vesoft-inc/nebula-kafka-connector/tree/v5.3.0)是用于从 Kafka 主题提取数据，并将其传输到悦数图数据库开发版的接收连接器。本文介绍如何设置 Kafka 连接器以实现从 Kafka 到悦数图数据库开发版的数据接收。

## 前提条件

- [Confluent Platform](https://docs.confluent.io/platform/current/overview.html) 已安装。
- 悦数图数据库开发版服务组正在运行中。更多信息，请参见[快速开始](https://dev-docs.yueshu.com.cn/v5.3.2/get-started/deploy/)。

## 步骤 1. 在悦数图数据库开发版中创建图

在从 Kafka 接收数据之前，需要在悦数图数据库开发版中创建一个用于存储数据的图。

在悦数图数据库开发版中，图是基于图类型创建的。图类型定义了图的数据结构。为确保数据正确接收，图类型必须与源数据的数据结构匹配。

本文使用[电影数据集](https://dev-docs.yueshu.com.cn/v5.3.2/database-gql-reference/overview/example-dataset/)作为示例，需要您在悦数图数据库开发版中创建一个名为`movie_type`的图类型和一个名为`movie`的图，如下所示：

````
'''
// 创建名为`movie_type`的图类型
CREATE GRAPH TYPE IF NOT EXISTS movie_type AS { 
    NODE Actor (LABEL Person {id INT64, name STRING, birthDate DATE, PRIMARY KEY (id)}) , 
    NODE Director (LABEL Person {id INT64, name STRING, birthDate DATE, PRIMARY KEY (id)}), 
    NODE `User` (LABEL Person {id INT64 PRIMARY KEY}), 
    NODE Movie (LABEL Movie {id INT64 PRIMARY KEY, name STRING}), 
    NODE Genre (LABEL Genre {id INT64 PRIMARY KEY, name STRING}), 
    EDGE Act (Actor)-[:Act]->(Movie), 
    EDGE Direct (Director)-[:Direct]->(Movie), 
    EDGE Watch (`User`)-[:Watch {rate FLOAT}]->(Movie), 
    EDGE WithGenre (Movie)-[:WithGenre]->(Genre) 
    }
'''

'''
// 基于图类型创建图
CREATE GRAPH IF NOT EXISTS movie TYPED movie_type
'''

````

关于图类型和图的更多信息，请参见 [CREATE GRAPH TYPE](https://dev-docs.yueshu.com.cn/v5.3.2/database-gql-reference/ddl/graph-type/create/) 和 [CREATE GRAPH](https://dev-docs.yueshu.com.cn/v5.3.2/database-gql-reference/ddl/graph/create/)。

## 步骤 2. 安装 Kafka 连接器

将 Kafka 连接器安装包解压缩为 JAR 包文件，并将 JAR 包文件放置到 Kafka 环境的目录`KAFKA_HOME/libs`中。

## 步骤 3. 准备配置文件

配置文件必须为`properties`格式，并包含以下信息：

- 悦数图数据库开发版服务组内 Graph 服务的连接信息。
- 从中提取数据的 Kafka 主题。
- Kafka 主题中的字段与图元素的属性之间的映射关系。

为了方便快速设置配置文件，您可以点击[此处](https://docs-cdn.nebula-graph.com.cn/files/5.x-utilities/kafka/sample_config.zip)下载电影数据集的示例配置文件。以下`connect-nebula-sink_actor.properties`文件是示例配置文件之一，用于从 Kafka 主题`actor`接收数据到图`movie`中的点类型`Actor`：

将`<current-password>`替换为登录账号的当前密码。若已修改密码，请使用修改后的密码。

````
name=local-nebula-sink
connector.class=com.vesoft.nebula.connector.sink.NebulaSinkConnector
tasks.max=1
topics=actor

nebula.graph.servers=127.0.0.1:9669
nebula.user=root
nebula.passwd=<current-password>
nebula.graph.name=movie
nebula.data.type=NODE
nebula.node.typeName=Actor
nebula.node.primaryKeys=id
kafka.node.primaryKeys=people_id
nebula.node.property.names=id,name,birthDate
kafka.node.property.names=people_id,name,birth_date
nebula.batchSize=50
nebula.sink.mode=INSERTREPLACE

nebula.request.timeout=5000
nebula.sin.partitions=10

key.converter=org.apache.kafka.connect.json.JsonConverter
value.converter=org.apache.kafka.connect.json.JsonConverter
key.converter.schemas.enable=false
value.converter.schemas.enable=false

````

请确保将以下参数替换为实际值：

- `nebula.graph.servers`：悦数图数据库开发版服务组中 Graph 服务的 IP 地址和端口号。
- `nebula.user`：连接 Graph 服务的用户名。
- `nebula.passwd`：登录账号的当前密码。若已修改密码，请使用修改后的密码。

有关配置参数的更多信息，请参见 [Kafka 连接器配置](https://dev-docs.yueshu.com.cn/v5.3.2/clients/kafka-connector/kafka-options/)。

## 步骤 4. 向 Kafka 主题生产数据

在向 Kafka 主题生产数据时，请确保用于生产数据的 JSON 文件中的字段与配置文件中定义的映射关系相匹配。

以下脚本演示了如何创建 Kafka 主题并将电影数据从 JSON 文件生产到这些主题中：

````
#!/bin/sh

# 创建 Kafka 主题
${KAFKA_HOME}/bin/kafka-topics.sh --create --topic actor --broker-list "127.0.0.1:9092" --replication-factor 1 --partitions 1
${KAFKA_HOME}/bin/kafka-topics.sh --create --topic director --broker-list "127.0.0.1:9092" --replication-factor 1 --partitions 1
${KAFKA_HOME}/bin/kafka-topics.sh --create --topic genre --broker-list "127.0.0.1:9092" --replication-factor 1 --partitions 1
${KAFKA_HOME}/bin/kafka-topics.sh --create --topic movie --broker-list "127.0.0.1:9092" --replication-factor 1 --partitions 1
${KAFKA_HOME}/bin/kafka-topics.sh --create --topic user --broker-list "127.0.0.1:9092" --replication-factor 1 --partitions 1
${KAFKA_HOME}/bin/kafka-topics.sh --create --topic act --broker-list "127.0.0.1:9092" --replication-factor 1 --partitions 1
${KAFKA_HOME}/bin/kafka-topics.sh --create --topic direct --broker-list "127.0.0.1:9092" --replication-factor 1 --partitions 1
${KAFKA_HOME}/bin/kafka-topics.sh --create --topic withGenre --broker-list "127.0.0.1:9092" --replication-factor 1 --partitions 1
${KAFKA_HOME}/bin/kafka-topics.sh --create --topic watched --broker-list "127.0.0.1:9092" --replication-factor 1 --partitions 1

# 将电影数据生产到 Kafka 主题中
cat movie_data/actor.json | ${KAFKA_HOME}/bin/kafka-console-producer.sh --broker-list "127.0.0.1:9092" --topic actor
cat movie_data/director.json | ${KAFKA_HOME}/bin/kafka-console-producer.sh --broker-list "127.0.0.1:9092" --topic director
cat movie_data/genre.json | ${KAFKA_HOME}/bin/kafka-console-producer.sh --broker-list "127.0.0.1:9092" --topic genre
cat movie_data/movie.json | ${KAFKA_HOME}/bin/kafka-console-producer.sh --broker-list "127.0.0.1:9092" --topic movie
cat movie_data/user.json | ${KAFKA_HOME}/bin/kafka-console-producer.sh --broker-list "127.0.0.1:9092" --topic user
cat movie_data/actor_act_movie.json | ${KAFKA_HOME}/bin/kafka-console-producer.sh --broker-list "127.0.0.1:9092" --topic act
cat movie_data/director_direct_movie.json | ${KAFKA_HOME}/bin/kafka-console-producer.sh --broker-list "127.0.0.1:9092" --topic direct
cat movie_data/movie_withgenre_genre.json | ${KAFKA_HOME}/bin/kafka-console-producer.sh --broker-list "127.0.0.1:9092" --topic withgenre
cat movie_data/user_watched_movies.json | ${KAFKA_HOME}/bin/kafka-console-producer.sh --broker-list "127.0.0.1:9092" --topic watched

````

您可以点击[此处](https://docs-cdn.nebula-graph.com.cn/files/5.x-utilities/kafka/movie_data.zip)下载上述脚本中使用的 JSON 文件。

除了 JSON，也支持将其他数据结构的数据（如 Struct 和 Map）生产到 Kafka 主题中。

## 步骤 5. 从 Kafka 主题接收数据到悦数图数据库开发版

将步骤 3 中准备的配置文件放置到目录`${KAFKA_HOME}/config`中，再运行以下命令将电影数据从 Kafka 主题接收至悦数图数据库开发版：

````
${KAFKA_HOME}/bin/connect-standalone.sh ${KAFKA_HOME}/config/standalone.properties ${KAFKA_HOME}/config/connect-nebula-sink_actor.properties
${KAFKA_HOME}/bin/connect-standalone.sh ${KAFKA_HOME}/config/standalone.properties ${KAFKA_HOME}/config/connect-nebula-sink_director.properties
${KAFKA_HOME}/bin/connect-standalone.sh ${KAFKA_HOME}/config/standalone.properties ${KAFKA_HOME}/config/connect-nebula-sink_user.properties
${KAFKA_HOME}/bin/connect-standalone.sh ${KAFKA_HOME}/config/standalone.properties ${KAFKA_HOME}/config/connect-nebula-sink_movie.properties
${KAFKA_HOME}/bin/connect-standalone.sh ${KAFKA_HOME}/config/standalone.properties ${KAFKA_HOME}/config/connect-nebula-sink_genre.properties
${KAFKA_HOME}/bin/connect-standalone.sh ${KAFKA_HOME}/config/standalone.properties ${KAFKA_HOME}/config/connect-nebula-sink_act.properties
${KAFKA_HOME}/bin/connect-standalone.sh ${KAFKA_HOME}/config/standalone.properties ${KAFKA_HOME}/config/connect-nebula-sink_direct.properties
${KAFKA_HOME}/bin/connect-standalone.sh ${KAFKA_HOME}/config/standalone.properties ${KAFKA_HOME}/config/connect-nebula-sink_watched.properties
${KAFKA_HOME}/bin/connect-standalone.sh ${KAFKA_HOME}/config/standalone.properties ${KAFKA_HOME}/config/connect-nebula-sink_withgenre.properties

````

## 步骤 6. 验证悦数图数据库开发版中的数据

要验证数据是否成功从 Kafka 主题接收至悦数图数据库开发版，可以连接到悦数图数据库开发版服务组并依次执行以下语句：

1. 

提交一个作业以生成图`movie`的统计信息。

````
SUBMIT JOB STATS movie

````

返回：

````
+--------+
| job_id |
+--------+
| 1      |
+--------+

````

2. 

显示图`movie`的统计信息。

````
USE movie SHOW STATS

````

如果数据成功接收，将看到类似以下的返回：

````
+------------+--------------+-----------+
| entry_name | element_type | total_num |
+------------+--------------+-----------+
| Edge Total | Edge         | 10175     |
| Node Total | Node         | 4589      |
| Act        | Edge         | 3732      |
| Direct     | Edge         | 249       |
| Watch      | Edge         | 3910      |
| WithGenre  | Edge         | 2284      |
| Actor      | Node         | 3312      |
| Director   | Node         | 183       |
| User       | Node         | 569       |
| Movie      | Node         | 245       |
| Genre      | Node         | 280       |
+------------+--------------+-----------+

````

## 更多信息

- [Kafka 连接器 GitHub 仓库](https://github.com/vesoft-inc/nebula-kafka-connector/tree/v5.3.0)
- [Kafka 连接器配置](https://dev-docs.yueshu.com.cn/v5.3.2/clients/kafka-connector/kafka-options/)