本文介绍如何创建Elasticsearch Sink Connector将数据从消息队列Kafka版实例的数据源Topic导出至阿里云Elasticsearch的索引。

前提条件

在导出数据前,请确保您已完成以下操作:

注意事项

  • 仅支持在同地域内,将数据从消息队列Kafka版实例的数据源Topic导出至函数计算,再由函数计算导出至Elasticsearch。Connector的限制说明,请参见使用限制
  • 该功能基于函数计算服务提供。函数计算为您提供了一定的免费额度,超额部分将产生费用,请以函数计算的计费规则为准。计费详情,请参见计费概述
  • 函数计算的函数调用支持日志查询。具体操作步骤,请参见配置并查看函数日志
  • 消息转储时,消息队列Kafka版中消息用UTF-8 String序列化,暂不支持二进制的数据格式。

创建并部署Elasticsearch Sink Connector

  1. 登录消息队列Kafka版控制台
  2. 在顶部菜单栏,选择地域。
  3. 在左侧导航栏,单击实例列表
  4. 实例列表页面,单击目标实例名称。
  5. 在左侧导航栏,单击Connector(公测组件)
  6. Connector(公测组件)页面,单击创建Connector
  7. 创建Connector配置向导中,完成以下操作。
    1. 基础信息页签的Connector名称文本框,输入Connector名称,然后将转储路径配置为从消息队列Kafka版转储到Elasticsearch,单击下一步
      注意 消息队列Kafka版会为您自动选中授权创建服务关联角色
      • 如果未创建服务关联角色,消息队列Kafka版会为您自动创建一个服务关联角色,以便您使用消息队列Kafka版导出数据至Elasticsearch的功能。
      • 如果已创建服务关联角色,消息队列Kafka版不会重复创建。
      关于该服务关联角色的更多信息,请参见服务关联角色
      参数 描述 示例值
      Connector名称 Connector的名称。命名规则:
      • 可以包含数字、小写英文字母和短划线(-),但不能以短划线(-)开头,长度限制为48个字符。
      • 同一个消息队列Kafka版实例内保持唯一。

      Connector的数据同步任务必须使用名称为connect-任务名称的Consumer Group。如果您未手动创建该Consumer Group,系统将为您自动创建。

      kafka-elasticsearch-sink
      转储路径 配置数据转储的源和目标。第一个下拉列表中选择数据源,第二个下拉列表中选择目标。 消息队列Kafka版转储到Elasticsearch
    2. 源实例配置页签,配置以下参数,然后单击下一步
      参数 描述 示例值
      数据源Topic 需要同步数据的Topic。 elasticsearch-test-input
      消费初始位置 开始消费的位置。取值:
      • latest:从最新位点开始消费。
      • earliest:从最初位点开始消费。
      latest
      消费线程并发数 数据源Topic的消费线程并发数。默认值为6。取值:
      • 6
      • 12
      6
    3. 目标实例配置 运行环境配置页签,配置目标实例相关参数。
      参数 描述 示例值
      ES实例ID 阿里云Elasticsearch实例ID。 es-cn-oew1o67x0000****
      接入地址 阿里云Elasticsearch实例的公网或私网地址。详细信息,请参见查看实例的基本信息 es-cn-oew1o67x0000****.elasticsearch.aliyuncs.com
      接入端口 访问阿里云Elasticsearch的公网或私网端口,取值如下:
      • 9200:基于HTTP或HTTPS。
      • 9300:基于TCP。

      详细信息,请参见查看实例的基本信息

      9300
      用户名 登录Kibana控制台的用户名,默认为elastic。您也可以创建自定义用户,创建步骤,请参见创建用户 elastic
      用户密码 登录Kibana控制台的密码。elastic用户的密码在创建实例时设定,如果忘记可重置。如需重置,请参见重置实例访问密码 ********
      索引 阿里云Elasticsearch的索引名称。 elastic_test
      说明
      • 用户名和用户密码会被用来初始化Elasticsearch对象,通过bulk投递消息,请确认账号对索引有写权限。
      • 用户名和用户密码是消息队列Kafka版创建任务时作为环境变量传递至函数计算的函数,任务创建成功后,消息队列Kafka版不保存相关信息。
    4. 目标实例配置 运行环境配置下方,配置运行环境,然后单击下一步
      参数 描述 示例值
      VPC ID 数据同步任务所在的VPC。默认为消息队列Kafka版实例所在的VPC,您无需填写。 vpc-bp1xpdnd3l***
      VSwitch ID 数据同步任务所在的交换机。该交换机必须与消息队列Kafka版实例处于同一VPC。默认为部署消息队列Kafka版实例时填写的交换机。 vsw-bp1d2jgg81***
      失败处理策略 消息发送失败后的错误处理。默认为log。取值:
      • log:继续订阅出现错误的Topic的分区,并打印错误日志。出现错误后,您可以通过Connector日志查看错误,并根据错误码查找解决方案,以进行自助排查。
        说明
      • fail:停止对出现错误的Topic的分区的订阅,并打印错误日志。出现错误后,您可以通过Connector日志查看错误,并根据错误的错误码查找解决方案,以进行自助排查。
        说明
        • 如何查看Connector日志,请参见查看Connector日志
        • 如何根据错误码查找解决方案,请参见错误码
        • 如需恢复对出现错误的Topic的分区的订阅,您需要提交提交工单联系消息队列Kafka版技术人员。
      log
      创建资源 选择自动创建或者选择手动创建并输入手动创建的Topic的名称。 自动创建
      Connector消费组 Connector使用的Consumer Group。该Consumer Group的名称建议以connect-cluster开头。 connect-cluster-kafka-elasticsearch-sink
      任务位点Topic 用于存储消费位点的Topic。
      • Topic名称:建议以connect-offset开头。
      • 分区数:Topic的分区数量必须大于1。
      • 存储引擎:Topic的存储引擎必须为Local存储。
      • cleanup.policy:Topic的日志清理策略必须为compact。
      connect-offset-kafka-elasticsearch-sink
      任务配置Topic 用于存储任务配置的Topic。
      • Topic名称:建议以connect-config开头。
      • 分区数:Topic的分区数量必须为1。
      • 存储引擎:Topic的存储引擎必须为Local存储。
      • cleanup.policy:Topic的日志清理策略必须为compact。
      connect-config-kafka-elasticsearch-sink
      任务状态Topic 用于存储任务状态的Topic。
      • Topic名称:建议以connect-status开头。
      • 分区数:Topic的分区数量建议为6。
      • 存储引擎:Topic的存储引擎必须为Local存储。
      • cleanup.policy:Topic的日志清理策略必须为compact。
      connect-status-kafka-elasticsearch-sink
      异常数据Topic 用于存储Sink的异常数据的Topic。该Topic可以和死信队列Topic为同一个Topic,以节省Topic资源。
      • Topic名称:建议以connect-error开头。
      • 分区数:Topic的分区数量建议为6。
      • 存储引擎:Topic的存储引擎可以为Local存储或云存储。
      connect-error-kafka-elasticsearch-sink
      死信队列Topic 用于存储Connect框架的异常数据的Topic。该Topic可以和异常数据Topic为同一个Topic,以节省Topic资源。
      • Topic名称:建议以connect-error开头。
      • 分区数:Topic的分区数量建议为6。
      • 存储引擎:Topic的存储引擎可以为Local存储或云存储。
      connect-error-kafka-elasticsearch-sink
    5. 预览/提交页签,确认Connector的配置,然后单击提交
  8. 在Connector列表,找到您刚创建的Connector,并在其操作列单击部署
    Connector状态切换至运行中,则说明部署成功。

配置函数服务

您在消息队列Kafka版控制台成功创建并部署Elasticsearch Sink Connector后,函数计算会自动为您创建给该Connector使用的函数服务,服务命名格式为kafka-service-<connector_name>-<随机String>

  1. Connector(公测组件)页面,找到目标Connector,在其右侧操作列,单击函数配置
    页面跳转至函数计算控制台。
  2. 在函数计算控制台,找到自动创建的函数服务,并配置其VPC和交换机信息。请确保该信息和您阿里云Elasticsearch相同。配置的具体步骤,请参见更新服务

发送测试消息

您可以向消息队列Kafka版的数据源Topic发送消息,测试数据能否被导出至阿里云Elasticsearch。

  1. Connector(公测组件)页面,找到目标Connector,在其右侧操作列,单击测试
  2. Topic管理页面,选择实例,找到数据源Topic,在其右侧操作列,单击发送消息
  3. 发送消息面板,发送测试消息。
    1. 分区文本框,输入0
    2. Message Key文本框,输入1
    3. Message Value文本框,输入1
    4. 单击发送

验证结果

消息队列Kafka版的数据源Topic发送消息后,登录Kibana控制台,执行GET /<index_name>/_search查看索引,验证数据导出结果。

消息队列Kafka版数据导出至Elasticsearch的格式示例如下:
{
  "took" : 8,
  "timed_out" : false,
  "_shards" : {
    "total" : 5,
    "successful" : 5,
    "skipped" : 0,
    "failed" : 0
  },
  "hits" : {
    "total" : {
      "value" : 1,
      "relation" : "eq"
    },
    "max_score" : 1.0,
    "hits" : [
      {
        "_index" : "product_****",
        "_type" : "_doc",
        "_id" : "TX3TZHgBfHNEDGoZ****",
        "_score" : 1.0,
        "_source" : {
          "msg_body" : {
            "key" : "test",
            "offset" : 2,
            "overflowFlag" : false,
            "partition" : 2,
            "timestamp" : 1616599282417,
            "topic" : "dv****",
            "value" : "test1",
            "valueSize" : 8
          },
          "doc_as_upsert" : true
        }
      }
    ]
  }
}