创建FC Sink Connector_创建Connector_Connector_用户指南

本文说明如何创建FC Sink Connector将数据从消息队列Kafka版导出至函数计算。

前提条件

在创建FC Sink Connector前，请确保您已完成以下操作：

根据网络类型购买并部署消息队列Kafka版实例。详情请参见VPC接入或公网+VPC接入。
为消息队列Kafka版实例开启Connector。详情请参见开启Connector。

在函数计算服务创建服务名称为guide-hello_world、函数名称为hello_world、运行环境为python的事件函数。创建完成后，将函数代码改成如下示例：

# -*- coding: utf-8 -*-
import logging

# To enable the initializer feature
# please implement the initializer function as below:
# def initializer(context):
#   logger = logging.getLogger()
#   logger.info('initializing')

def handler(event, context):
  logger = logging.getLogger()
  logger.info('hello world:' + bytes.decode(event))
  return 'hello world:' + bytes.decode(event)

详情请参见使用控制台创建函数。

操作流程

使用FC Sink Connector将数据从消息队列Kafka版的Topic导出至函数计算的操作流程如下：

授予消息队列Kafka版访问函数计算的权限
创建FC Sink Connector所需的消息队列Kafka版资源
1. 创建Consumer Group
2. 创建Topic
创建FC Sink Connector
创建FC Sink Connector
结果验证
1. 发送消息
2. 查看函数日志

创建自定义权限策略

创建访问函数计算的自定义权限策略。

登录访问控制控制台。
在左侧导航栏，选择权限管理 > 权限策略管理。
在权限策略管理页面，单击创建权限策略。
在新建自定义权限策略页面，创建自定义权限策略。
1. 在策略名称文本框，输入KafkaConnectorFcAccess。
2. 在配置模式区域，选择脚本配置。
3. 在策略内容区域，输入自定义权限策略脚本。
  访问函数计算的自定义权限策略脚本示例如下：
```
{
    "Version": "1",
    "Statement": [
        {
            "Action": [
                "fc:InvokeFunction",
                "fc:GetFunction"
            ],
            "Resource": "*",
            "Effect": "Allow"
        }
    ]
}
```
4. 单击确定。

创建RAM角色

由于RAM角色不支持直接选择消息队列Kafka版作为受信服务，您在创建RAM角色时，需要选择任意支持的服务作为受信服务。RAM角色创建后，手工修改信任策略。

在左侧导航栏，单击RAM角色管理。
在RAM角色管理页面，单击创建RAM角色。
在创建RAM角色向导页面，创建RAM角色。
1. 在当前可信实体类型区域，选择阿里云服务，单击下一步。
2. 在角色类型区域，选择普通服务角色，在角色名称文本框，输入AliyunKafkaConnectorRole，从选择受信服务列表，选择函数计算，然后单击完成。
在RAM角色管理页面，找到AliyunKafkaConnectorRole，单击AliyunKafkaConnectorRole。
在AliyunKafkaConnectorRole页面，单击信任策略管理页签，单击修改信任策略。
在修改信任策略对话框，将脚本中fc替换为alikafka，单击确定。

添加权限

为创建的RAM角色授予访问函数计算的权限。

在左侧导航栏，单击RAM角色管理。
在RAM角色管理页面，找到AliyunKafkaConnectorRole，在其右侧操作列，单击添加权限。
在添加权限对话框，添加KafkaConnectorFcAccess权限。
1. 在选择权限区域，选择自定义权限策略。
2. 在权限策略名称列表，找到KafkaConnectorFcAccess，单击KafkaConnectorFcAccess。
3. 单击确定。
4. 单击完成。

创建Consumer Group

授予消息队列Kafka版的RAM角色访问函数计算的权限后，创建FC Sink Connector数据同步任务所需的Consumer Group。

登录消息队列Kafka版控制台。
在顶部菜单栏，选择地域。
在左侧导航栏，单击Consumer Group管理。
在Consumer Group管理页面，选择实例，单击创建Consumer Group。
在创建Consumer Group对话框，创建任务消费组。
1. 在Consumer Group文本框，输入connect-kafka-fc-sink。
2. 在备注文本框，输入connect-kafka-fc-sink。
3. 单击创建。

重复步骤4和步骤5，创建Connector消费组。

表 1. Consumer Group参数说明
Consumer Group类型	参数	描述	示例值
Connector消费组	Consumer Group	Consumer Group的名称。建议以connect-cluster开头。	connect-cluster-kafka-fc-sink

Consumer Group管理页面显示创建的Consumer Group。 connect-cluster-kafka-fc-sink-result

创建Topic

创建FC Sink Connector数据同步任务所需的Topic。

在左侧导航栏，单击Topic管理。
在Topic管理页面，单击创建Topic。
在创建Topic对话框，创建数据源Topic。
1. 在Topic文本框，输入fc-test-input。
2. 在备注文本框，输入fc-test-input。
3. 单击创建。

重复步骤2和步骤3，创建以下Topic：


Topic类型	参数	描述	示例值
任务位点Topic	Topic	Topic的名称。建议以connect-offset开头。	connect-offset-fc-sink
	分区数	Topic的分区数量。分区数量必须大于1。	12
	存储引擎	Topic的存储引擎。存储引擎必须为Local存储。	Local存储
	cleanup.policy	Topic的日志清理策略。日志清理策略必须为compact。	compact
任务配置Topic	Topic	Topic的名称。建议以connect-config开头。	connect-config-fc-sink
	分区数	Topic的分区数量。分区数量必须为1。	1
	存储引擎	Topic的存储引擎。存储引擎必须为Local存储。	Local存储
	cleanup.policy	Topic的日志清理策略。日志清理策略必须为compact。	compact
任务状态Topic	Topic	Topic的名称。建议以connect-status开头。	connect-status-fc-sink
	分区数	Topic的分区数量。分区数量建议为6。	6
	存储引擎	Topic的存储引擎。存储引擎必须为Local存储。	Local存储
	cleanup.policy	Topic的日志清理策略。日志清理策略必须为compact。	compact
死信队列Topic	Topic	Topic的名称。	fc_dead_letter_error
	分区数	Topic的分区数量。分区数量建议为6。	6
	存储引擎	Topic的存储引擎。存储引擎可以为Local存储或云存储。	云存储
异常数据Topic	Topic	Topic的名称。	fc_runtime_error
	分区数	Topic的分区数量。分区数量建议为6。	6
	存储引擎	Topic的存储引擎。存储引擎可以为Local存储或云存储。	云存储

说明用于存储异常数据的死信队列Topic和异常数据Topic可以使用同一个Topic。

Topic管理页面显示创建的Topic。 fc_topic_result