本文说明如何创建MaxCompute Sink Connector将数据从消息队列Kafka版实例的数据源Topic导出至MaxCompute的表。
前提条件
- 为消息队列Kafka版实例开启Connector。更多信息,请参见开启Connector。
- 为消息队列Kafka版实例创建数据源Topic。更多信息,请参见步骤一:创建Topic。
本文以名称为maxcompute-test-input的Topic为例。
- 通过MaxCompute客户端创建表。更多信息,请参见创建和查看表。
本文以名称为connector_test的项目下名称为test_kafka的表为例。该表的建表语句如下:
CREATE TABLE IF NOT EXISTS test_kafka(topic STRING,partition BIGINT,offset BIGINT,key STRING,value STRING) PARTITIONED by (pt STRING);
操作流程
使用MaxCompute Sink Connector将数据从消息队列Kafka版实例的数据源Topic导出至MaxCompute的表操作流程如下:
- 授予消息队列Kafka版访问MaxCompute的权限
- 可选:创建MaxCompute Sink Connector依赖的Topic和Consumer Group
如果您不需要自定义Topic和Consumer Group的名称,您可以直接跳过该步骤,在下一步骤选择自动创建。
注意 部分MaxCompute Sink Connector依赖的Topic的存储引擎必须为Local存储,大版本为0.10.2的消息队列Kafka版实例不支持手动创建Local存储的Topic,只支持自动创建。 - 创建并部署MaxCompute Sink Connector
- 结果验证
创建RAM角色
由于RAM角色不支持直接选择消息队列Kafka版作为受信服务,您在创建RAM角色时,需要选择任意支持的服务作为受信服务。RAM角色创建后,手工修改信任策略。
添加权限
为使Connector将消息同步到MaxCompute表,您需要为创建的RAM角色至少授予以下权限:
客体 | 操作 | 描述 |
---|---|---|
Project | CreateInstance | 在项目中创建实例。 |
Table | Describe | 读取表的元信息。 |
Table | Alter | 修改表的元信息或添加删除分区。 |
Table | Update | 覆盖或添加表的数据。 |
关于以上权限的详细说明以及授权操作,请参见授权。
为本文创建的AliyunKafkaMaxComputeUser1添加权限的示例步骤如下:
创建MaxCompute Sink Connector依赖的Topic
您可以在消息队列Kafka版控制台手动创建MaxCompute Sink Connector依赖的5个Topic。
创建MaxCompute Sink Connector依赖的Consumer Group
您可以在消息队列Kafka版控制台手动创建MaxCompute Sink Connector依赖的2个Consumer Group。
创建并部署MaxCompute Sink Connector
创建并部署用于将数据从消息队列Kafka版同步至MaxCompute的MaxCompute Sink Connector。
- 登录消息队列Kafka版控制台。
- 在顶部菜单栏,选择地域。
- 在Connector页面,选择实例,单击创建Connector。
- 在创建Connector面板,完成以下操作。
- 在创建Connector面板,单击部署。
发送测试消息
部署MaxCompute Sink Connector后,您可以向消息队列Kafka版的数据源Topic发送消息,测试数据能否被同步至MaxCompute。
- 在Connector页面,找到目标Connector,在其右侧操作列,单击测试。
- 在Topic管理页面,选择实例,找到maxcompute-test-input,在其右侧操作列,选择
> 单击发送消息。
- 在发送消息对话框,发送测试消息。
- 在分区文本框,输入0。
- 在Message Key文本框,输入1。
- 在Message Value文本框,输入1。
- 单击发送。
查看表数据
向消息队列Kafka版的数据源Topic发送消息后,在MaxCompute客户端查看表数据,验证是否收到消息。
查看本文写入的test_kafka的示例步骤如下: