ApsaraMQ for Kafka：建立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
      自动添加数据表	大量新增資料庫中的其他表。當建立的新表匹配成功時，也可被識別並同步資料。 格式為Regex。例如，輸入.*，表示匹配資料庫中的所有表。	.*
      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"
   }