文档中心

云搜索服务

数据集成

通过 Logstash 读取消息队列 Kafka 版数据写入云搜索服务

本文介绍如何通过 Logstash 管道，实现读取消息队列 Kafka 版 Topic 中的消息数据，并写入到云搜索服务索引中。

背景信息

消息队列 Kafka 版是一款基于 Apache Kafka 构建的分布式消息中间件服务。具备高吞吐、高可扩展性等特性，提供流式数据的发布/订阅和多副本存储机制，广泛应用于日志压缩收集、流式数据处理、消息解耦、流量削峰去谷等应用场景。您可以使用 Logstash Kafka 连接器连接消息队列 Kafka 版，通过 Logstash 管道将消息队列 Kafka 版 Topic 中的消息数据消费到下游的大数据组件或者数据仓库。

流程介绍

准备数据源消息队列 Kafka 版 Topic。

您需要在消息队列 Kafka 版控制台创建一个实例，然后创建一个 Topic。

还需要获取实例的访问地址和 Topic 名称，以便在 Logstash 管道中填入信息实现与消息队列 Kafka 版的连通。

准备数据目的云搜索服务索引。

您需要在云搜索服务控制台创建实例并获取实例的访问地址。

创建 Logstash 管道。

当您准备好数据源和数据目的后，便可以在云搜索服务控制台创建 Logstash 管道。

注意事项

为保证网络访问安全，本文所使用的云产品服务均使用内网访问方式，因此要求消息队列 Kafka 版实例与云搜索服务实例均处于相同地域的同一个 VPC 内。

您可以在创建云资源前，先创建私有网络。具体操作，请参见创建私有网络和创建子网。

步骤一：准备消息队列 Kafka 版 Topic

通过 Logstash 管道消费消息队列 Kafka 版消息前，首先需要先为消息队列 Kafka 版完成如下准备工作：

创建实例并获取实例 ID。具体操作，请参见创建实例和查看实例详情。

创建 Topic 并获取 Topic 名称。具体操作，请参见创建 Topic 和查看 Topic 详情。

说明

实例 ID：在读取消息队列 Kafka 版 Topic 中的消息数据时，作为 Kafka SASL 用户名。

Topic 名称：在生产消息的 Logstash 管道中，用于指定消息写入的目标主题。

步骤二：准备数据目标端云搜索服务索引

通过 Logstash 管道消费消息队列 Kafka 版消息前，需要准备一个数据写入的目的源，本文使用的是云搜索服务索引。请按照以下步骤创建实例，并获取实例的访问地址。

创建云搜索服务实例。具体操作，请参见创建实例。

根据消息队列 Kafka 版的数据格式，确定索引 Schema，并在 Dev Tools 中创建索引。具体操作，请参见 Dev Tools。示例语句如下。

PUT test_orders
{
"mappings": {
"properties": {
"k_field": {
"type": "keyword"
},
"i_field": {
"type": "integer"
}
}
}
}

步骤三：创建 Logstash 管道

在步骤二的云搜索服务实例的同一地域和 VPC 中创建 Logstash 实例。具体操作，请参见创建 Logstash 实例。

创建 Logstash 管道，引用 Input-kafka 和 Output-elasticsearch/opensearch 模板，并根据实际情况修改模板中的内容。具体操作，请参见创建管道。配置示例如下。

input {
kafka {
# 协议类型 PLAINTEXT
bootstrap_servers => ["kafka******.ivolces.com:****"] # 消息队列 Kafka 版实例的连接地址
client_id => "logstash_client"
group_id => "logstash_consumer"
auto_offset_reset => "latest"
consumer_threads => 5
decorate_events => true
topics => ["***_logstash_test"]
# 主题中存储 JSON 格式内容，会解析后按字段写入云搜索服务索引
codec => json {
ecs_compatibility => "disabled"
}
}
}
output {
opensearch {
# 目标实例 endpoint
hosts => "https://elasticsearch-o-dev-******.escloud-dev.ivolces.com:****"# 云搜索实例访问链接
# 目标实例 basicAuth 鉴权信息，如果存在
user => "admin"
password => "********"
# 和源索引名保持一致
index => "test_orders"
document_id => "%{id}"
# 是否仅用 `hosts` 配置项数组中的 HTTPS 链接进行请求
ssl => false
# 当使用 HTTPS 链接时，是否验证云搜索服务端证书
ssl_certificate_verification => false
# ssl_certificate_verification 为 true 时，验证服务端 CA 的证书路径，.cer 或 .pem格式
# cacert => "ca证书路径"
}
}

保存并部署管道，等待 Logstash 实例状态再次变成运行中时，在日志中查看管道是否正常运行。具体操作，请参见查询日志。

结果验证

在 Dev Tools 中查询执行索引的数据验证写入结果。具体操作，请参见 Dev Tools。示例命令如下。

示例输出如下。

最近更新时间：2026.07.31 15:55:02

这个页面对您有帮助吗？

有用

无用