将消息从 Apache Kafka 导入到像 Databend 这样的云数据仓库可能是一项艰巨的任务。bend-ingest-kafka 工具简化了这一过程，使您能够轻松地将 Kafka 消息导入到 Databend 表中。
什么是 bend-ingest-kafka？
bend-ingest-kafka 工具是一个专门为从 Apache Kafka 到 Databend 进行消息加载的命令行工具，支持以下模式：
-
JSON 模式：此模式解析 JSON 消息并将每个字段加载到目标表的相应列中。它将 Kafka 消息转换为与表模式直接映射的结构化格式。
-
RAW 模式：在此模式下，该工具会创建一个包含
uuid、koffset、kpartition、raw_data、record_metadata和add_time等列的表。Kafka 消息存储在raw_data列中，并将其他元数据注入到其他列中，从而保留原始消息格式并提供有用的上下文。
有关 bend-ingest-kafka 的更多详细信息，请查看 README。要安装 bend-ingest-kafka，请运行以下命令：
go install github.com/databendcloud/bend-ingest-kafka@latest
如何使用 bend-ingest-kafka
在这里中，我们将安装一个本地 Kafka 实例，创建一个主题，并生成一些示例消息。然后，我们将分别以 JSON 和 RAW 模式将这些消息加载到 Databend Cloud 中。
-
安装 Kafka
在开始之前，请确保您已运行 Kafka 并创建了一个主题。以下是使用 Docker 的快速部署：
使用 Docker 运行 Kafka 容器：
docker run -d \
- name kafka \
-p 9092:9092 \
apache/kafka:latest
创建一个名为 “test-topic” 的主题并生成一些消息：
docker exec - workdir /opt/kafka/bin/ -it kafka sh
/opt/kafka/bin $ ./kafka-topics.sh - bootstrap-server localhost:9092 - create - topic test-topic
Created topic test-topic.
/opt/kafka/bin $ ./kafka-console-producer.sh - bootstrap-server localhost:9092 - topic test-topic
>{"id": 1, "name": "Alice", "age": 30}
>{"id": 2, "name": "Bob", "age": 25}
-
在 Databend Cloud 创建目标表
在 Databend Cloud 中创建一个数据库和表。这张表将作为 JSON 模式加载的目标，因此请确保其模式与之前生成的消息匹配。
CREATE DATABASE doc;
CREATE TABLE databend_topic (
id INT NOT NULL,
name VARCHAR NOT NULL,
age INT NOT NULL
) ENGINE=FUSE;
-
使用 JSON 模式加载
使用以下命令开始将消息从 Kafka 加载到 Databend Cloud：
要获取您的 DSN，请登录到 Databend Cloud 并点击“连接” 。
bend-ingest-kafka \
- kafka-bootstrap-servers="localhost:9092" \
- kafka-topic="test-topic" \
- databend-dsn="<your-dsn>" \
- databend-table="doc.databend_topic" \
- data-format="json"
您将在终端中看到类似这样的消息：
INFO[0000] Starting worker worker-0
2024/08/27 14:31:18 ingest 1 rows (0.900730 rows/s), 38 bytes (34.227748 bytes/s)
WARN[0095] Failed to read message from Kafka: context deadline exceeded kafka_batch_reader=ReadBatch
2024/08/27 14:32:19 ingest 1 rows (1.206118 rows/s), 36 bytes (43.420260 bytes/s)
验证 Kafka 消息是否已成功加载到 Databend Cloud 中的目标表，并且每个字段都正确映射到其指定的列中：
-
使用 RAW 模式加载
使用以下命令开始将消息从 Kafka 加载到 Databend Cloud：
bend-ingest-kafka \
- kafka-bootstrap-servers="localhost:9092" \
- kafka-topic="test-topic" \
- databend-dsn="<your-dsn>" \
- is-json-transform=false
您将在终端中看到类似这样的消息：
INFO[0000] Starting worker worker-0
WARN[0077] Failed to read message from Kafka: context deadline exceeded kafka_batch_reader=ReadBatch
2024/08/27 15:12:56 ingest 2 rows (1.978622 rows/s), 579 bytes (572.811134 bytes/s)
验证 Kafka 消息已成功加载到 Databend Cloud 中，并自动创建了一张表：
