将云消息队列Kafka版接入阿里云Elasticsearch

登录云消息队列 Kafka 版控制台。

在概览页面的资源分布区域，选择地域。

在实例列表页面，单击目标实例名称。

在实例详情页面的接入点信息页签，获取实例的VPC环境接入点。

登录云消息队列 Kafka 版控制台。

在概览页面的资源分布区域，选择地域。

在实例列表页面，单击目标实例名称。

在左侧导航栏，单击Topic 管理。

在Topic 管理页面，单击创建 Topic。

在创建 Topic面板，设置Topic属性，然后单击确定。

登录云消息队列 Kafka 版控制台。

在概览页面的资源分布区域，选择地域。

在实例列表页面，单击目标实例名称。

在左侧导航栏，单击Topic 管理。

在Topic 管理页面，找到目标Topic，在其操作列中，选择更多 > 体验发送消息。

在快速体验消息收发面板，发送测试消息。

• 发送方式选择控制台。

  1. 在消息 Key文本框中输入消息的Key值，例如demo。

  2. 在消息内容文本框输入测试的消息内容，例如 {"key": "test"}。

  3. 设置发送到指定分区，选择是否指定分区。

     • 单击是，在分区 ID文本框中输入分区的ID，例如0。如果您需查询分区的ID，请参见查看分区状态。

     • 单击否，不指定分区。

  4. 根据界面提示信息，通过SDK订阅消息，或者执行Docker命令订阅消息。

• 发送方式选择Docker，运行Docker容器。

  1. 执行运行 Docker 容器生产示例消息区域的Docker命令，发送消息。

  2. 执行发送后如何消费消息？区域的Docker命令，订阅消息。

• 发送方式选择SDK，根据您的业务需求，选择需要的语言或者框架的SDK以及接入方式，通过SDK体验消息收发。

登录云消息队列 Kafka 版控制台。

在概览页面的资源分布区域，选择地域。

在实例列表页面，单击目标实例名称。

在左侧导航栏，单击Group 管理。

在Group 管理页面，单击创建 Group。

在创建 Group面板的Group ID文本框输入Group的名称，在描述文本框简要描述Group，并给Group添加标签，单击确定。

登录Kibana控制台。

在Kibana控制台的左侧导航栏，单击Dev Tools。

执行以下命令创建索引。

PUT /elastic_test
{}

登录阿里云Elasticsearch控制台的Logstash实例页面。

在顶部菜单栏，选择地域。

在Logstash实例页面，单击创建的实例。

在左侧导航栏，单击管道管理。

在管道列表区域，单击创建管道。

在Config配置中，输入配置。

input {
    kafka {
    bootstrap_servers => ["alikafka-pre-cn-zv**********-1-vpc.alikafka.aliyuncs.com:9092,alikafka-pre-cn-zv**********-2-vpc.alikafka.aliyuncs.com:9092,alikafka-pre-cn-zv**********-3-vpc.alikafka.aliyuncs.com:9092"]
    group_id => "elastic_group"
    topics => ["elastic_test"]
    consumer_threads => 12
    decorate_events => true
    }
}
output {
    elasticsearch {
    hosts => ["http://es-cn-o40xxxxxxxxxxxxwm.elasticsearch.aliyuncs.com:9200"]
    index => "elastic_test"
    password => "XXX"
    user => "elastic"
    }
}

在管道参数配置中，输入配置信息，然后单击保存并部署。

在提示对话框，单击确认。

登录Kibana控制台。

在Kibana控制台的左侧导航栏，单击Dev Tools图标。

执行以下命令搜索数据。

GET /elastic_test/_search
{}