如何创建Tablestore Sink Connector_云消息队列 Kafka 版(Kafka)-阿里云帮助中心

本文介绍如何创建使用Tablestore Sink Connector，您可通过Tablestore Sink Connector将数据从云消息队列 Kafka 版实例的数据源Topic导出至Tablestore的表中。

前提条件

创建资源及授权策略，请参见创建前提。
创建Connector时生成的函数所在的服务角色需要手动添加AliyunOTSFullAccess权限策略，赋予此服务角色管理表格存储服务（OTS）的权限。具体操作，请参见授予函数计算访问其他云服务的权限。

本文以实例为ots-sink和数据表名为ots_sink_table的资源为例，创建数据表时设置pk1、pk2两个主键。

在创建任务页面，设置任务名称和描述，配置以下信息，然后单击保存。

任务创建

在Source（源）配置向导，选择数据提供方为消息队列 Kafka 版，设置以下参数，然后单击下一步。

参数	说明	示例
地域	选择云消息队列 Kafka 版源实例所在的地域。	华北2（北京）
kafka 实例	选择生产云消息队列 Kafka 版消息的源实例。	MQ_INST_115964845466****_ByBeUp3p
Topic	选择生产云消息队列 Kafka 版消息的Topic。	topic
Group ID	选择源实例的消费组名称。请使用独立的消费组来创建事件源，不要和已有的业务混用消费组，以免影响已有的消息收发。	GID_http_1
消费位点	选择开始消费消息的位点。	最新位点
网络配置	选择路由消息的网络类型。	默认网络
专有网络VPC	选择VPC ID。当网络配置设置为自建公网时需要设置此参数。	vpc-bp17fapfdj0dwzjkd****
交换机	选择vSwitch ID。当网络配置设置为自建公网时需要设置此参数。	vsw-bp1gbjhj53hdjdkg****
安全组	选择安全组。当网络配置设置为自建公网时需要设置此参数。	alikafka_pre-cn-7mz2****
批量推送条数	调用函数发送的最大批量消息条数，当积压的消息数量到达设定值时才会发送请求，取值范围为 [1,10000]。	100
批量推送间隔（单位：秒）	调用函数的间隔时间，系统每到间隔时间点会将消息聚合后发给函数计算，取值范围为[0,15]，单位为秒。0秒表示无等待时间，直接投递。	3

在Sink（目标）配置向导，选择服务类型为表格存储TableStore，配置以下参数。

参数	说明	示例
实例名称	已创建的Tablestore实例名称。	ost-sink
目标表	已创建的Tablestore数据表名称。	ost_sink_table
主键	Tablestore主键值生成方式，需定义每个主键内容的提取规则，提取规则通过JsonPath语法表达。	例如，主键名称为id，数值提取规则配置为`$.data.value.id`
属性列	属性列生成方式，需定义每个属性列内容的提取规则，提取规则通过JsonPath语法表达。	例如，属性列名称为key，数值提取规则配置为`$.data.value.name`
写入模式	数据写入Tablestore的方式。 put：覆盖写入行。 update：更新写入行。	put
网络配置	专有网络：通过专有网络VPC将Kafka消息投递到Tablestore。公网：通过公网将Kafka消息投递到Tablestore。	公网
VPC	选择VPC ID。仅当网络配置为专有网络时需配置此参数。	vpc-bp17fapfdj0dwzjkd****
交换机	选择vSwitch ID。仅当网络配置为专有网络时需配置此参数。	vsw-bp1gbjhj53hdjdkg****
安全组	选择安全组。仅当网络配置为专有网络时需配置此参数。	test_group