Kafka 连接器指南¶

Kafka 连接器是用于从 Kafka 主题提取数据，并将其传输到悦数图数据库开发版的接收连接器。本文介绍如何设置 Kafka 连接器以实现从 Kafka 到悦数图数据库开发版的数据接收。

前提条件¶

Confluent Platform 已安装。
悦数图数据库开发版服务组正在运行中。更多信息，请参见快速开始。

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

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

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

本文使用电影数据集作为示例，需要您在悦数图数据库开发版中创建一个名为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 和 CREATE GRAPH。

步骤 2. 安装 Kafka 连接器¶

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

步骤 3. 准备配置文件¶

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

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

为了方便快速设置配置文件，您可以点击此处下载电影数据集的示例配置文件。以下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 连接器配置。

步骤 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

您可以点击此处下载上述脚本中使用的 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 主题接收至悦数图数据库开发版，可以连接到悦数图数据库开发版服务组并依次执行以下语句：

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

SUBMIT JOB STATS movie

返回：

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

显示图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       |
+------------+--------------+-----------+