云消息队列 Kafka 版：创建SLS Sink Connector

步骤一：创建目标服务资源

• 创建日志项目（Project）。详细步骤，请参见创建项目Project。

• 创建日志库（Logstore）。详细步骤，请参见创建Logstore。

本文以Project为guide-sls-sink-project和Logstore为guide-sls-sink-logstore的资源为例。

步骤二：创建SLS Sink Connector并启动

1. 登录云消息队列 Kafka 版控制台，在概览页面的资源分布区域，选择地域。

2. 在左侧导航栏，选择Connector 生态集成 > 任务列表。

3. 在任务列表页面，单击创建任务。

   • 任务创建

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

     参数	说明	示例
     地域	选择云消息队列 Kafka 版源实例所在的地域。	华北2（北京）
     kafka 实例	选择生产云消息队列 Kafka 版消息的源实例。	alikafka_post-cn-jte3****
     Topic	选择生产云消息队列 Kafka 版消息的Topic。	topic
     Group ID	数据源所在的云消息队列 Kafka 版实例的Group ID。 快速创建：新建一个Group ID资源， 使用已有：填写一个已经的创建好的GroupID资源。	GID_http_1
     消费位点	选择开始消费消息的位点。	最新位点
     网络配置	选择路由消息的网络类型。	基础网络
     专有网络VPC	选择VPC ID。当网络配置设置为自建公网时需要设置此参数。	vpc-bp17fapfdj0dwzjkd****
     交换机	选择vSwitch ID。当网络配置设置为自建公网时需要设置此参数。	vsw-bp1gbjhj53hdjdkg****
     安全组	选择安全组。当网络配置设置为自建公网时需要设置此参数。	alikafka_pre-cn-7mz2****
     批量推送条数	调用函数发送的最大批量消息条数，当积压的消息数量到达设定值时才会发送请求，取值范围为 [1,10000]。	100
     批量推送间隔（单位：秒）	调用函数的间隔时间，系统每到间隔时间点会将消息聚合后发给函数计算，取值范围为[0,15]，单位为秒。0秒表示无等待时间，直接投递。	3

     2. 在Filtering（过滤）配置向导，设置数据模式内容过滤发送的请求。更多信息，请参见事件模式。

     3. 在Transform（转换）配置向导，设置数据清洗，实现分割、映射、富化及动态路由等繁杂数据加工能力。更多信息，请参见数据清洗。

     4. 在Sink（目标）配置向导，选择服务类型为日志服务，配置以下参数，单击保存。

     表 1. 目标（日志服务SLS）

     参数	说明	示例
     日志项目	已创建的日志服务SLS Project。	guide-sls-sink-project
     日志库	已创建的日志服务SLS Logstore。	guide-sls-sink-logstore
     日志主题	SLS日志Topic生成方式。	空
     角色配置	授权事件总线EventBridge写入SLS的权限，如果没有角色，可按照控制台提示授权创建新角色即可。	sls_eb

4. 返回任务列表页面，找到创建好的任务，在其右侧操作列，单击启用。

5. 在提示对话框，阅读提示信息，然后单击确认。

   启用任务后，会有30秒~60秒的延迟时间，您可以在任务列表页面的状态栏查看启动进度。

步骤三：测试SLS Sink Connector

1. 在任务列表页面，在SLS Sink Connector任务的事件源列单击源Topic。

2. 在Topic详情页面，单击体验发送消息。

3. 在快速体验消息收发面板，按照下图配置消息内容，然后单击确定。

   

4. 在任务列表页面，在SLS Sink Connector任务的事件目标列单击目标Project。

5. 在日志库页面，查看日志内容。