本文介绍如何创建OSS Sink Connector将数据从云消息队列 Kafka 版实例的数据源Topic导出至对象存储OSS。

前提条件

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

注意事项

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

创建并部署OSS Sink Connector

  1. 登录云消息队列 Kafka 版控制台。
  2. 在概览页面的资源分布区域，选择地域。
  3. 在左侧导航栏，单击Connector 任务列表。
  4. 在Connector 任务列表页面，从选择实例的下拉列表选择Connector所属的实例，然后单击创建 Connector。
  5. 在创建 Connector配置向导面页面，完成以下操作。
    1. 在配置基本信息页签，按需配置以下参数，然后单击下一步。
      重要 云消息队列 Kafka 版会为您自动选中授权创建服务关联角色。
      • 如果未创建服务关联角色，云消息队列 Kafka 版会为您自动创建一个服务关联角色，以便您使用云消息队列 Kafka 版导出数据至对象存储的功能。
      • 如果已创建服务关联角色，云消息队列 Kafka 版不会重复创建。
      关于该服务关联角色的更多信息，请参见服务关联角色。
      参数描述示例值
      名称Connector的名称。命名规则：
      • 可以包含数字、小写英文字母和短划线（-），但不能以短划线（-）开头，长度限制为48个字符。
      • 同一个云消息队列 Kafka 版实例内保持唯一。

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

      		kafka-oss-sink
      实例默认配置为实例的名称与实例ID。demo alikafka_post-cn-st21p8vj****
    2. 在配置源服务页签，选择数据源为消息队列Kafka版，并配置以下参数，然后单击下一步。
      参数描述示例值
      数据源 Topic需要同步数据的Topic。oss-test-input
      消费线程并发数数据源Topic的消费线程并发数。默认值为6。取值说明如下：
      • 1
      • 2
      • 3
      • 6
      • 12
      		6
      消费初始位置开始消费的位置。取值说明如下：
      • 最早位点：从最初位点开始消费。
      • 最近位点：从最新位点开始消费。
      		最早位点
      VPC ID数据同步任务所在的VPC。单击配置运行环境显示该参数。默认为云消息队列 Kafka 版实例所在的VPC，您无需填写。vpc-bp1xpdnd3l***
      vSwitch ID数据同步任务所在的交换机。单击配置运行环境显示该参数。该交换机必须与云消息队列 Kafka 版实例处于同一VPC。默认为部署云消息队列 Kafka 版实例时填写的交换机。vsw-bp1d2jgg81***
      失败处理消息发送失败后，是否继续订阅出现错误的Topic的分区。单击配置运行环境显示该参数。取值说明如下。
      • 继续订阅：继续订阅出现错误的Topic的分区，并打印错误日志。
      • 停止订阅：停止订阅出现错误的Topic的分区，并打印错误日志
      说明
      		继续订阅
      创建资源方式选择创建Connector所依赖的Topic与Group的方式。单击配置运行环境显示该参数。
      • 自动创建
      • 手动创建
      		自动创建
      Connector 消费组Connector使用的Group。单击配置运行环境显示该参数。该Group的名称建议以connect-cluster开头。connect-cluster-kafka-oss-sink
      任务位点 Topic用于存储消费位点的Topic。单击配置运行环境显示该参数。
      • Topic：建议以connect-offset开头。
      • 分区数：Topic的分区数量必须大于1。
      • 存储引擎：Topic的存储引擎必须为Local存储。
        说明 仅专业版实例支持在创建Topic时选择存储引擎为Local存储，标准版暂不支持。
      • cleanup.policy：Topic的日志清理策略必须为compact。
      		connect-offset-kafka-oss-sink
      任务配置 Topic用于存储任务配置的Topic。单击配置运行环境显示该参数。
      • Topic：建议以connect-config开头。
      • 分区数：Topic的分区数量必须为1。
      • 存储引擎：Topic的存储引擎必须为Local存储。
        说明 仅专业版实例支持在创建Topic时选择存储引擎为Local存储，标准版暂不支持。
      • cleanup.policy：Topic的日志清理策略必须为compact。
      		connect-config-kafka-oss-sink
      任务状态 Topic用于存储任务状态的Topic。单击配置运行环境显示该参数。
      • Topic：建议以connect-status开头。
      • 分区数：Topic的分区数量建议为6。
      • 存储引擎：Topic的存储引擎必须为Local存储。
        说明 仅专业版实例支持在创建Topic时选择存储引擎为Local存储，标准版暂不支持。
      • cleanup.policy：Topic的日志清理策略必须为compact。
      		connect-status-kafka-oss-sink
      死信队列 Topic用于存储Connect框架的异常数据的Topic。单击配置运行环境显示该参数。该Topic可以和异常数据Topic为同一个Topic，以节省Topic资源。
      • Topic：建议以connect-error开头。
      • 分区数：Topic的分区数量建议为6。
      • 存储引擎：Topic的存储引擎可以为Local存储或云存储。
        说明 仅专业版实例支持在创建Topic时选择存储引擎为Local存储，标准版暂不支持。
      		connect-error-kafka-oss-sink
      异常数据 Topic用于存储Sink的异常数据的Topic。单击配置运行环境显示该参数。该Topic可以和死信队列Topic为同一个Topic，以节省Topic资源。
      • Topic：建议以connect-error开头。
      • 分区数：Topic的分区数量建议为6。
      • 存储引擎：Topic的存储引擎可以为Local存储或云存储。
        说明 仅专业版实例支持在创建Topic时选择存储引擎为Local存储，标准版暂不支持。
      		connect-error-kafka-oss-sink
    3. 在配置目标服务页签，选择目标服务为对象存储，并配置以下参数，然后单击创建。
      参数描述示例值
      Bucket 名称对象存储Bucket的名称。bucket_test
      Access Key阿里云账号的AccessKey ID。LTAI4GG2RGAjppjK********
      Secret Key阿里云账号的AccessKey Secret。WbGPVb5rrecVw3SQvEPw6R********

      请确保您使用的AccessKey ID所对应的账号已被授予以下最小权限：

      {
    "Version": "1",
    "Statement": [
        {
            "Action": [
                "oss:GetObject",
                "oss:PutObject"
            ],
            "Resource": "*",
            "Effect": "Allow"
        }
    ]
}
      说明

      AccessKey ID和AccessKey Secret是云消息队列 Kafka 版创建任务时作为环境变量传递至对象存储的数据，任务创建成功后，云消息队列 Kafka 版不保存AccessKey ID和AccessKey Secret信息。

      创建完成后，在Connector 任务列表页面，查看创建的Connector 。
  6. 创建完成后，在Connector 任务列表页面，找到创建的Connector ，单击其操作列的部署。

发送测试消息

您可以向云消息队列 Kafka 版的数据源Topic发送消息，测试数据能否被导出至对象存储。

  1. 在Connector 任务列表页面，找到目标Connector，在其右侧操作列，单击测试。
  2. 在发送消息面板，发送测试消息。
    • 发送方式选择控制台。
      1. 在消息 Key文本框中输入消息的Key值，例如demo。
      2. 在消息内容文本框输入测试的消息内容，例如 {"key": "test"}。
      3. 设置发送到指定分区，选择是否指定分区。
        • 单击是，在分区 ID文本框中输入分区的ID，例如0。如果您需查询分区的ID，请参见查看分区状态。
        • 单击否，不指定分区。
    • 发送方式选择Docker，执行运行 Docker 容器生产示例消息区域的Docker命令，发送消息。
    • 发送方式选择SDK，根据您的业务需求，选择需要的语言或者框架的SDK以及接入方式，通过SDK发送消息。

验证结果

向云消息队列 Kafka 版的数据源Topic发送消息后，查看OSS文件管理，验证数据导出结果。更多信息，请参见文件概览。

文件管理中显示新导出的文件。

files
云消息队列 Kafka 版数据导出至对象存储的格式示例如下：
[
    {
        "key":"123",
        "offset":4,
        "overflowFlag":true,
        "partition":0,
        "timestamp":1603779578478,
        "topic":"Test",
        "value":"1",
        "valueSize":272687
    }
]

更多操作

您可以按需对该Connector所依赖的函数计算资源进行配置。

在Connector 任务列表页面，找到创建的Connector，单击其操作列的更多 > 配置函数。
页面跳转至函数计算控制台，您可以按需配置函数资源。