本文介绍如何通过创建AnalyticDB Sink Connector，将数据从消息队列Kafka版实例的数据源Topic通过函数计算服务导出至云原生数据仓库AnalyticDB MySQL版或云原生数据仓库AnalyticDB PostgreSQL版。

前提条件

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

注意事项

  • 仅支持在同地域内，将数据从消息队列Kafka版实例的数据源Topic导出至函数计算，再由函数计算导出至云原生数据仓库AnalyticDB MySQL版或云原生数据仓库AnalyticDB PostgreSQL版。关于Connector的限制说明，请参见使用限制。
  • 该功能基于函数计算服务提供。函数计算为您提供了一定的免费额度，超额部分将产生费用，请以函数计算的计费规则为准。计费详情，请参见计费概述。
  • 函数计算的函数调用支持日志查询，以便您迅速排查问题。具体操作步骤，请参见配置日志。
  • 消息转储时，消息队列Kafka版中消息用UTF-8 String序列化，暂不支持二进制的数据格式。
  • 如果AnalyticDB Sink Connector接入点是私网接入点，函数计算运行环境默认无法访问，为确保网络畅通，需在函数计算控制台为函数服务配置与云原生数据仓库一致的VPC和vSwitch信息。更多信息，请参见更新服务。
  • 创建Connector时，消息队列Kafka版会为您自动创建服务关联角色。
    • 如果未创建服务关联角色，消息队列Kafka版会为您自动创建一个服务关联角色，以便您使用消息队列Kafka版导出数据至表格存储的功能。
    • 如果已创建服务关联角色，消息队列Kafka版不会重复创建。
    关于服务关联角色的更多信息，请参见服务关联角色。

操作流程

使用AnalyticDB Sink Connector将数据从消息队列Kafka版实例的数据源Topic导出至云原生数据仓库操作流程如下：

  1. 可选：创建AnalyticDB Sink Connector依赖的Topic和Group

    如果您不需要自定义Topic和Group，您可以直接跳过该步骤，在下一步骤选择自动创建。

    重要 部分AnalyticDB Sink Connector依赖的Topic的存储引擎必须为Local存储，大版本为0.10.2的消息队列Kafka版实例不支持手动创建Local存储的Topic，只支持自动创建。
    1. 创建AnalyticDB Sink Connector依赖的Topic
    2. 创建AnalyticDB Sink Connector依赖的Group
  2. 创建并部署AnalyticDB Sink Connector
  3. 服务配置
    1. 配置函数服务
    2. 配置云原生数据库
  4. 结果验证
    1. 发送测试消息
    2. 验证结果

创建AnalyticDB Sink Connector依赖的Topic

您可以在消息队列Kafka版控制台手动创建AnalyticDB Sink Connector依赖的5个Topic，包括：任务位点Topic、任务配置Topic、任务状态Topic、死信队列Topic以及异常数据Topic。每个Topic所需要满足的分区数与存储引擎会有差异，具体信息，请参见配置源服务参数列表。

  1. 登录消息队列Kafka版控制台。
  2. 在概览页面的资源分布区域，选择地域。
    重要 Topic需要在应用程序所在的地域（即所部署的ECS的所在地域）进行创建。Topic不能跨地域使用。例如Topic创建在华北2（北京）这个地域，那么消息生产端和消费端也必须运行在华北2（北京）的ECS。
  3. 在实例列表页面，单击目标实例名称。
  4. 在左侧导航栏，单击Topic 管理。
  5. 在Topic 管理页面，单击创建 Topic。
  6. 在创建 Topic面板，设置Topic属性，然后单击确定。
    创建Topic
    参数 说明 示例
    名称 Topic名称。 demo
    描述 Topic的简单描述。 demo test
    分区数 Topic的分区数量。 12
    存储引擎
    说明 当前仅专业版实例支持选择存储引擎类型，标准版暂不支持，默认选择为云存储类型。
    		Topic消息的存储引擎。

    消息队列Kafka版支持以下两种存储引擎。

    • 云存储：底层接入阿里云云盘，具有低时延、高性能、持久性、高可靠等特点，采用分布式3副本机制。实例的规格类型为标准版（高写版）时，存储引擎只能为云存储。
    • Local 存储：使用原生Kafka的ISR复制算法，采用分布式3副本机制。
    		云存储
    消息类型 Topic消息的类型。
    • 普通消息：默认情况下，保证相同Key的消息分布在同一个分区中，且分区内消息按照发送顺序存储。集群中出现机器宕机时，可能会造成消息乱序。当存储引擎选择云存储时，默认选择普通消息。
    • 分区顺序消息：默认情况下，保证相同Key的消息分布在同一个分区中，且分区内消息按照发送顺序存储。集群中出现机器宕机时，仍然保证分区内按照发送顺序存储。但是会出现部分分区发送消息失败，等到分区恢复后即可恢复正常。当存储引擎选择Local 存储时，默认选择分区顺序消息。
    		普通消息
    日志清理策略 Topic日志的清理策略。

    当存储引擎选择Local 存储（当前仅专业版实例支持选择存储引擎类型为Local存储，标准版暂不支持）时，需要配置日志清理策略。

    消息队列Kafka版支持以下两种日志清理策略。

    • Delete：默认的消息清理策略。在磁盘容量充足的情况下，保留在最长保留时间范围内的消息；在磁盘容量不足时（一般磁盘使用率超过85%视为不足），将提前删除旧消息，以保证服务可用性。
    • Compact：使用Kafka Log Compaction日志清理策略。Log Compaction清理策略保证相同Key的消息，最新的value值一定会被保留。主要适用于系统宕机后恢复状态，系统重启后重新加载缓存等场景。例如，在使用Kafka Connect或Confluent Schema Registry时，需要使用Kafka Compact Topic存储系统状态信息或配置信息。
      重要 Compact Topic一般只用在某些生态组件中，例如Kafka Connect或Confluent Schema Registry，其他情况的消息收发请勿为Topic设置该属性。具体信息，请参见消息队列Kafka版Demo库。
    		Compact
    标签 Topic的标签。 demo
    创建完成后，在Topic 管理页面的列表中显示已创建的Topic。

创建AnalyticDB Sink Connector依赖的Group

您可以在消息队列Kafka版控制台手动创建AnalyticDB Sink Connector数据同步任务使用的Group。该Group的名称必须为connect-任务名称，具体信息，请参见配置源服务参数列表。

  1. 登录消息队列Kafka版控制台。
  2. 在概览页面的资源分布区域，选择地域。
  3. 在实例列表页面，单击目标实例名称。
  4. 在左侧导航栏，单击Group 管理。
  5. 在Group 管理页面，单击创建 Group。
  6. 在创建 Group面板的Group ID文本框输入Group的名称，在描述文本框简要描述Group，并给Group添加标签，单击确定。
    创建完成后，在Group 管理页面的列表中显示已创建的Group。

创建并部署AnalyticDB Sink Connector

  1. 登录消息队列Kafka版控制台。
  2. 在概览页面的资源分布区域，选择地域。
  3. 在左侧导航栏，单击Connector 任务列表。
  4. 在Connector 任务列表页面，从选择实例的下拉列表选择Connector所属的实例，然后单击创建 Connector。
  5. 在创建 Connector配置向导页面，完成以下操作。
    1. 在配置基本信息页签，按需配置以下参数，然后单击下一步。
      参数 描述 示例值
      名称 Connector的名称。命名规则：
      • 可以包含数字、小写英文字母和短划线（-），但不能以短划线（-）开头，长度限制为48个字符。
      • 同一个消息队列Kafka版实例内保持唯一。

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

      		kafka-adb-sink
      实例 默认配置为实例的名称与实例ID。 demo alikafka_post-cn-st21p8vj****
    2. 在配置源服务页签，选择数据源为消息队列Kafka版，并配置以下参数，然后单击下一步。
      表 1. 配置源服务参数列表
      参数 描述 示例值
      数据源 Topic 需要同步数据的Topic。 adb-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所依赖的Topic与Group的方式。单击配置运行环境显示该参数。
      • 自动创建
      • 手动创建
      		 自动创建
      Connector 消费组 Connector的数据同步任务使用的Group。单击配置运行环境显示该参数。该Group的名称必须为connect-任务名称。 connect-kafka-adb-sink
      任务位点 Topic 用于存储消费位点的Topic。单击配置运行环境显示该参数。
      • Topic：建议以connect-offset开头。
      • 分区数：Topic的分区数量必须大于1。
      • 存储引擎：Topic的存储引擎必须为Local存储。
      • cleanup.policy：Topic的日志清理策略必须为compact。
      		 connect-offset-kafka-adb-sink
      任务配置 Topic 用于存储任务配置的Topic。单击配置运行环境显示该参数。
      • Topic：建议以connect-config开头。
      • 分区数：Topic的分区数量必须为1。
      • 存储引擎：Topic的存储引擎必须为Local存储。
      • cleanup.policy：Topic的日志清理策略必须为compact。
      		 connect-config-kafka-adb-sink
      任务状态 Topic 用于存储任务状态的Topic。单击配置运行环境显示该参数。
      • Topic：建议以connect-status开头。
      • 分区数：Topic的分区数量建议为6。
      • 存储引擎：Topic的存储引擎必须为Local存储。
      • cleanup.policy：Topic的日志清理策略必须为compact。
      		 connect-status-kafka-adb-sink
      死信队列 Topic 用于存储Connect框架的异常数据的Topic。单击配置运行环境显示该参数。该Topic可以和异常数据Topic为同一个Topic，以节省Topic资源。
      • Topic：建议以connect-error开头。
      • 分区数：Topic的分区数量建议为6。
      • 存储引擎：Topic的存储引擎可以为Local存储或云存储。
      		 connect-error-kafka-adb-sink
      异常数据 Topic 用于存储Sink的异常数据的Topic。单击配置运行环境显示该参数。该Topic可以和死信队列Topic为同一个Topic，以节省Topic资源。
      • Topic：建议以connect-error开头。
      • 分区数：Topic的分区数量建议为6。
      • 存储引擎：Topic的存储引擎可以为Local存储或云存储。
      		 connect-error-kafka-adb-sink
    3. 在配置目标服务页签，选择目标服务为云原生数据仓库，并配置以下参数，然后单击创建。
      参数 描述 示例值
      实例类型 云原生数据仓库实例类型。支持MySQL版和PostgreSQL版。 MySQL版
      AnalyticDB 实例 ID 阿里云云原生数据仓库AnalyticDB MySQL版或云原生数据仓库AnalyticDB PostgreSQL版实例ID。 am-bp139yqk8u1ik****
      数据库名 阿里云云原生数据仓库实例的数据库名称。 adb_demo
      表名 云原生数据仓库中存储消息表名称。 user
      数据库用户名 连接云原生数据仓库实例数据库导入数据的数据库用户名。 adbmysql
      数据库密码 连接云原生数据仓库实例数据库导入数据的数据库密码。用户的密码在创建实例时设定，如果忘记可重置。 ********
      说明 用户名和用户密码是消息队列Kafka版创建任务时作为环境变量传递至函数计算的函数，任务创建成功后，消息队列Kafka版不保存相关信息。
      创建完成后，在Connector 任务列表页面，查看创建的Connector 。
  6. 创建完成后，在Connector 任务列表页面，找到创建的Connector ，单击其操作列的部署。

配置函数服务

您在消息队列Kafka版控制台成功创建并部署AnalyticDB Sink Connector后，函数计算会自动为您创建给该Connector使用的函数服务和函数，服务命名格式为kafka-service-<connector_name>-<随机String>，函数命名格式为fc-adb-<随机String>。

  1. 在Connector 任务列表页面，找到目标Connector，在其右侧操作列，单击函数配置。
    页面跳转至函数计算控制台。
  2. 在函数计算控制台，找到自动创建的函数服务，并配置其VPC和交换机信息。配置的具体步骤，请参见更新服务。

配置云原生数据库

您在配置完函数计算服务后，需要在云原生数据仓库控制台将函数计算服务所属的网段加入白名单。所属网段可以在专有网络管理控制台的交换机页面，函数计算服务对应的VPC和交换机所在行查看。

发送测试消息

您可以向消息队列Kafka版的数据源Topic发送消息，测试数据能否被导出至阿里云云原生数据仓库AnalyticDB MySQL版或云原生数据仓库AnalyticDB PostgreSQL版。
说明 消息内容（Value）格式需为JSON格式，Value将通过JSON解析为K-V形式，其中K对应云原生数据仓库的数据库中的字段名，V对应该字段插入的数据，因此消息队列Kafka版发送的消息内容中的每个K在数据库中需要有对应的相同名称的字段名。字段名可以在云原生数据仓库AnalyticDB MySQL控制台或云原生数据仓库AnalyticDB PostgreSQL版控制台连接数据库，在数据库表中查看。
  1. 在Connector 任务列表页面，找到目标Connector，在其右侧操作列，单击测试。
  2. 在发送消息面板，发送测试消息。
    • 发送方式选择控制台。
      1. 在消息 Key文本框中输入消息的Key值，例如demo。
      2. 在消息内容文本框输入测试的消息内容，例如 {"key": "test"}。
      3. 设置发送到指定分区，选择是否指定分区。
        • 单击是，在分区 ID文本框中输入分区的ID，例如0。如果您需查询分区的ID，请参见查看分区状态。
        • 单击否，不指定分区。
    • 发送方式选择Docker，执行运行 Docker 容器生产示例消息区域的Docker命令，发送消息。
    • 发送方式选择SDK，根据您的业务需求，选择需要的语言或者框架的SDK以及接入方式，通过SDK发送消息。

验证结果

向消息队列Kafka版的数据源Topic发送消息后，登录云原生数据仓库AnalyticDB MySQL控制台或云原生数据仓库AnalyticDB PostgreSQL版控制台，连接数据库，进入数据管理DMS 5.0的SQL窗口界面，找到对应实例的表，验证数据导出结果。

消息队列Kafka版数据导出至云原生数据仓库AnalyticDB MySQL版示例如下：ADB-Connector-Result