将数据从 AutoMQ for Kafka 导入 Databend
Databend是使用 Rust 研发、开源的、完全面向云架构、基于对象存储构建的新一代云原生数据仓库，为企业提供湖仓一体化、计算和存储分离的大数据分析平台。
本文将介绍如何通过 bend-ingest-kafka 将数据从 AutoMQ for Kafka 导入 Databend。
info
本文中提及的 AutoMQ Kafka 术语，均特指安托盟丘（杭州）科技有限公司通过 GitHub AutoMQ 组织下开源的 automq-for-kafka 项目。
环境准备
准备 Databend Cloud 以及测试数据
首先前往 Databend Cloud 开启你的 Warehouse ，并在 worksheet 中创建数据库库和测试表。
create database automq_db;create table users ( id bigint NOT NULL, name string NOT NULL, ts timestamp, status string)
准备 AutoMQ Kafka 环境和测试数据
参考 部署 AutoMQ 到 AWS▸ 部署好 AutoMQ Kafka 集群，确保 AutoMQ Kafka 与 Databend Cloud 之间保持网络连通。
在AutoMQ Kafka中快速创建一个名为 example_topic 的主题并向其中写入一条测试 JSON 数据，可以通过以下步骤实现。
创建Topic
使用 Apache Kafka 命令行工具来创建主题。你需要有 Kafka 环境的访问权限，并且确保 Kafka 服务正在运行。以下是创建主题的命令：
./kafka-topics.sh --create --topic exampleto_topic --bootstrap-server 10.0.96.4:9092 --partitions 1 --replication-factor 1
注意：执行命令时，需要将 topic 和 bootstarp-server 替换为实际使用的 Kafka 服务器地址。
创建 topic 之后可以用以下命令检查 topic 创建的结果。
./kafka-topics.sh --describe example_topic --bootstrap-server 10.0.96.4:9092
生成测试数据
生成一条简单的 JSON 格式的测试数据，和前文的表需要对应。
{ "id":1, "name":"Test User", "ts":"2023-11-10T12:00:00", "status":"active"}
写入测试数据
使用 Kafka 的命令行工具或者编程方式将测试数据写入到 example_topic。以下是使用命令行工具的一个示例：
echo '{"id": 1, "name": "测试用户", "ts": "2023-11-10T12:00:00", "status": "active"}' | sh kafka-console-producer.sh --broker-list 10.0.96.4:9092 --topic example_topic
使用如下命令可以查看刚写入的 topic 数据：
sh kafka-console-consumer.sh --bootstrap-server 10.0.96.4:9092 --topic example_topic --from-beginning
创建 bend-ingest-databend job
bend-ingest-kafka 能够监控 kafka 并将数据批量写入 Databend Table。
部署 bend-ingest-kafka 之后，即可开启数据导入 job。
bend-ingest-kafka --kafka-bootstrap-servers="localhost:9094" --kafka-topic="example_topic" --kafka-consumer-group="Consumer Group" --databend-dsn="https://cloudapp:password@host:443" --databend-table="automq_db.users" --data-format="json" --batch-size=5 --batch-max-interval=30s
note
将 kafka_broker_list 替换为实际使用的 Kafka 服务器地址。
参数说明
databend-dsn
Databend Cloud 提供的连接到 warehouse 的 DSN，可以参考该文档 获取。
batch-size
bend-ingest-kafka 会积攒到 batch-size 条数据再触发一次数据同步。
验证数据导入
到 Databend Cloud worksheet 中查询 automq_db.users 表，可以看到数据已经从 AutoMq 同步到 Databend Table。
END
关于我们
AutoMQ 是一家专业的消息队列和流存储软件服务供应商。AutoMQ 开源的 AutoMQ Kafka 和 AutoMQ RocketMQ 基于云对 Apache Kafka、Apache RocketMQ 消息引擎进行重新设计与实现，在充分利用云上的竞价实例、对象存储等服务的基础上，兑现了云设施的规模化红利，带来了下一代更稳定、高效的消息引擎。此外，AutoMQ 推出的 RocketMQ Copilot 专家系统也重新定义了 RocketMQ 消息运维的新范式，赋能消息运维人员更好的管理消息集群。
关注我们，一起学习更多云原生干货
