创建OSS Sink Connector将数据从Kafka导出至OSS_云消息队列 Kafka 版(Kafka)-阿里云帮助中心

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

前提条件

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

为云消息队列 Kafka 版实例开启Connector。更多信息，请参见开启Connector。
为云消息队列 Kafka 版实例创建数据源Topic。更多信息，请参见步骤一：创建Topic。
在OSS管理控制台创建存储空间。更多信息，请参见控制台创建存储空间。
开通函数计算服务。更多信息，请参见开通函数计算服务。

注意事项

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

创建并部署OSS Sink Connector

登录云消息队列 Kafka 版控制台。
在概览页面的资源分布区域，选择地域。
在左侧导航栏，单击Connector 任务列表。
在Connector 任务列表页面，从选择实例的下拉列表选择Connector所属的实例，然后单击创建 Connector。

在创建 Connector配置向导面页面，完成以下操作。

在配置基本信息页签，按需配置以下参数，然后单击下一步。

重要

云消息队列 Kafka 版会为您自动选中授权创建服务关联角色。

如果未创建服务关联角色，云消息队列 Kafka 版会为您自动创建一个服务关联角色，以便您使用云消息队列 Kafka 版导出数据至对象存储的功能。
如果已创建服务关联角色，云消息队列 Kafka 版不会重复创建。

关于该服务关联角色的更多信息，请参见服务关联角色。

参数

描述

示例值

名称

Connector的名称。命名规则：

可以包含数字、小写英文字母和短划线（-），但不能以短划线（-）开头，长度限制为48个字符。
同一个云消息队列 Kafka 版实例内保持唯一。

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

kafka-oss-sink

实例

默认配置为实例的名称与实例ID。

demo alikafka_post-cn-st21p8vj****

在配置源服务页签，选择数据源为消息队列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相关操作。如何根据错误码查找解决方案，请参见错误码。	继续订阅
创建资源方式	选择创建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

在配置目标服务页签，选择目标服务为对象存储，并配置以下参数，然后单击创建。
参数
描述
示例值
Bucket 名称
对象存储Bucket的名称。
bucket_test
Access Key
阿里云账号的AccessKey ID。
yourAccessKeyID
Secret Key
阿里云账号的AccessKey Secret。
yourAccessKeySecret
请确保您使用的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 。

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

发送测试消息

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

在Connector 任务列表页面，找到目标Connector，在其右侧操作列，单击测试。
在发送消息面板，发送测试消息。
- 发送方式选择控制台。
  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
    }
]