Kafka connector guide

The Kafka connector is a sink connector that pulls data from Kafka topics and streams it to NebulaGraph Database Developer Edition. This topic introduces how to set up the Kafka connector to sink data from Kafka into NebulaGraph Database Developer Edition.

Prerequisites

• Confluent Platform is installed.
• A NebulaGraph Database Developer Edition service group is running. For more information, see Getting started with NebulaGraph Database Developer Edition.

Step 1. Create a graph in NebulaGraph Database Developer Edition

Before you sink data from Kafka into NebulaGraph Database Developer Edition, you need to create a graph in NebulaGraph Database Developer Edition to store the data.

In NebulaGraph Database Developer Edition, graphs are created based on graph types, which define the data structure of the graphs. To ensure that your data is written correctly, the graph type must match the data structure of the source data.

This topic uses the movie dataset as an example and requires you to create a graph type named movie_type and a graph named movie as follows in NebulaGraph Database Developer Edition:

'''
// Create a graph type named `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 a graph named based on graph type
CREATE GRAPH IF NOT EXISTS movie TYPED movie_type
'''

For more information about graph types and graphs, see CREATE GRAPH TYPE and CREATE GRAPH.

Step 2. Install the Kafka connector

Decompress the Kafka connector package into a JAR file and place the JAR file into your Kafka environment: KAFKA_HOME/libs.

Step 3. Prepare the configuration files

The configuration files must be in the properties format and contain the following information:

• The connection information for the Graph service in a NebulaGraph Database Developer Edition service group.
• The Kafka topics from which the connector pulls data.
• Mapping between the fields in the Kafka topics and the properties of the graph elements in NebulaGraph Database Developer Edition.

To help you quickly set up the configuration files, you can click here to download the sample configuration files for the movie dataset. The following connect-nebula-sink_actor.properties is one of the sample configuration files that you can use to sink data from the Kafka topic actor into the node type Actor in the graph movie:

Replace <current-password> with the current password for the specified account. If the password has changed, use the new 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

Make sure to replace the following parameters with the actual values:

• nebula.graph.servers: The IP address and port number of the Graph service in your NebulaGraph Database Developer Edition service group.
• nebula.user: The username for connecting to the Graph service.
• nebula.passwd: The current password for the account used to connect. If the password has changed, use the new password.

For more information about the configuration parameters, see Kafka connector configurations.

Step 4. Produce data to Kafka topics

When producing data, ensure the fields in the JSON files produced to Kafka topics match the mapping defined in the configuration files.

The following script demonstrates how to create Kafka topics and produce movie data from JSON files to these topics:

#!/bin/sh

# Create Kafka topics
${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

# Produce movie data in the JSON format into Kafka topics
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

You can click here to download the JSON files used in the above script.

Apart from JSON, you can also produce data to Kafka topics using other data structures, such as structs and maps.

Step 5. Sink data from Kafka topics into NebulaGraph Database Developer Edition

Place the configuration files prepared in Step 3 into the directory ${KAFKA_HOME}/config and run the following commands to sink movie data from Kafka topics into NebulaGraph Database Developer Edition:

${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

Step 6. Verify the data in NebulaGraph Database Developer Edition

To verify that the data is successfully written from Kafka topics into NebulaGraph Database Developer Edition, you can connect to the NebulaGraph Database Developer Edition service group and execute the following statements in sequence:

1. Submit a job to generate statistics for the movie graph.

   SUBMIT JOB STATS movie
   

   Return:

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

2. Show the statistics for the movie graph.

   USE movie SHOW STATS
   

   The returned result should be similar to the following if the data is written successfully:

   +------------+--------------+-----------+
   | 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       |
   +------------+--------------+-----------+
   

More information

• Kafka connector GitHub repository
• Kafka connector configurations