创建MySQL Source Connector

注意事项

• 地域说明

  • 如果数据源和目标实例位于不同地域，请确保您使用的账号拥有云企业网实例，且云企业网实例已挂载数据源和目标实例所在的VPC，并配置好流量带宽完成网络打通。

    否则，可能会新建云企业网实例，并将目标实例和独享资源组ECS全部挂载到云企业网实例来打通网络。这样的云企业网实例没有配置带宽，所以带宽流量很小，可能导致创建同步任务过程中的网络访问出错，或者同步任务创建成功后，在运行过程中出错。

  • 如果数据源和目标实例位于同一地域，创建数据同步任务会自动在其中一个实例所在VPC创建ENI，并绑定到独享资源组ECS上，以打通网络。

• DataWorks独享资源组说明

  • DataWorks的每个独享资源组可以运行最多3个同步任务。创建数据同步任务时，如果DataWorks发现您的账号名下有资源组的历史购买记录，并且运行的同步任务少于3个，将使用已有资源组运行新建的同步任务。

  • DataWorks的每个独享资源组最多绑定两个VPC的ENI。如果DataWorks发现已购买的资源组绑定的ENI与需要新绑定的ENI有网段冲突，或者其他技术限制，导致使用已购买的资源组无法创建出同步任务，此时，即使已有的资源组运行的同步任务少于3个，也将新建资源组确保同步任务能够顺利创建。

创建并部署MySQL Source Connector

1. 登录云消息队列 Kafka 版控制台。

2. 在概览页面的资源分布区域，选择地域。

3. 在左侧导航栏，单击Connector 任务列表。

4. 在Connector 任务列表页面，从选择实例的下拉列表选择Connector所属的实例，然后单击创建 Connector。

5. 在创建 Connector配置向导中，完成以下操作。

   1. 在配置基本信息页签的名称文本框，输入Connector名称，然后单击下一步。

      参数	描述	示例值
      名称	Connector的名称。命名规则： 可以包含数字、小写英文字母和短划线（-），但不能以短划线（-）开头，长度限制为48个字符。 同一个云消息队列 Kafka 版实例内保持唯一。 Connector的数据同步任务必须使用名称为connect-任务名称的Group。如果您未手动创建该Group，系统将为您自动创建。	kafka-source-mysql
      实例	默认配置为实例的名称与实例ID。	demo alikafka_post-cn-st21p8vj****

   2. 在配置源服务页签，选择数据源为云数据库RDS MySQL版，配置以下参数，然后单击下一步。

      参数	描述	示例值
      RDS 实例所在地域	从下拉列表中，选择阿里云数据库RDS MySQL版实例所在的地域。	华南1（深圳）
      云数据库 RDS 实例 ID	需要同步数据的阿里云数据库RDS MySQL版的实例ID。	rm-wz91w3vk6owmz****
      数据库名称	需要同步的阿里云数据库RDS MySQL版实例数据库的名称。	mysql-to-kafka
      数据库账号	需要同步的阿里云数据库RDS MySQL版实例数据库账号。	mysql_to_kafka
      数据库账号密码	需要同步的阿里云数据库RDS MySQL版实例数据库账号的密码。	无
      数据库表	需要同步的阿里云数据库RDS MySQL版实例数据库表的名称，多个表名以英文逗号（,）分隔。 数据库表和目标Topic是一对一的关系。	mysql_tbl
      自动添加数据表	批量添加数据库中的其他表。当创建的新表匹配成功时，也可被识别并同步数据。 格式为正则表达式。例如，输入.*，表示匹配数据库中的所有表。	.*
      Topic 前缀	阿里云数据库RDS MySQL版数据库表同步到云消息队列 Kafka 版的Topic的命名前缀，请确保前缀全局唯一。	mysql

      重要

      请确保阿里云数据库RDS MySQL版数据库账号有以下最小权限：

      • SELECT

      • REPLICATION SLAVE

      • REPLICATION CLIENT

      授权命令示例：

      GRANT SELECT, REPLICATION SLAVE, REPLICATION CLIENT ON *.* TO '同步账号'@'%'; //授予数据库账号的SELECT、REPLICATION SLAVE和REPLICATION CLIENT权限。

   3. 在配置目标服务页签，显示数据将同步到目标云消息队列 Kafka 版实例，确认信息无误后，单击创建。

6. 创建完成后，在Connector 任务列表页面，找到创建的Connector ，单击其操作列的部署。

   在Connector 任务列表页面，您可以看到创建的任务状态为运行中，则说明任务创建成功。

   说明

   如果创建失败，请再次检查本文前提条件中的操作是否已全部完成。

   如需配置同步任务，单击其操作列的任务配置，跳转至DataWorks控制台完成操作。

验证结果

1. 向阿里云数据库RDS MySQL版数据库表插入数据。

   示例如下。

   INSERT INTO mysql_tbl
       (mysql_title, mysql_author, submission_date)
       VALUES
       ("mysql2kafka", "tester", NOW())

   更多SQL语句，请参见常用语句。

2. 使用云消息队列 Kafka 版提供的消息查询功能，验证数据能否被导出至云消息队列 Kafka 版目标Topic。

   查询的具体步骤，请参见消息查询。

   云数据库RDS MySQL版数据库表导出至云消息队列 Kafka 版Topic的数据示例如下。消息结构及各字段含义，请参见附录：消息格式。

   {
       "schema":{
           "dataColumn":[
               {
                   "name":"mysql_id",
                   "type":"LONG"
               },
               {
                   "name":"mysql_title",
                   "type":"STRING"
               },
               {
                   "name":"mysql_author",
                   "type":"STRING"
               },
               {
                   "name":"submission_date",
                   "type":"DATE"
               }
           ],
           "primaryKey":[
               "mysql_id"
           ],
           "source":{
               "dbType":"MySQL",
               "dbName":"mysql_to_kafka",
               "tableName":"mysql_tbl"
           }
       },
       "payload":{
           "before":null,
           "after":{
               "dataColumn":{
                   "mysql_title":"mysql2kafka",
                   "mysql_author":"tester",
                   "submission_date":1614700800000
               }
           },
           "sequenceId":"1614748790461000000",
           "timestamp":{
               "eventTime":1614748870000,
               "systemTime":1614748870925,
               "checkpointTime":1614748870000
           },
           "op":"INSERT",
           "ddl":null
       },
       "version":"0.0.1"
   }