ApsaraMQ for Kafka:MySQL ソースコネクタの作成

注意事項

• リージョン
  • データソースとデスティネーションの Message Queue for Apache Kafka インスタンスは、同じリージョンにデプロイされていない場合があります。この場合、Alibaba Cloud アカウント内に Cloud Enterprise Network (CEN) インスタンスがあり、データソースと Message Queue for Apache Kafka インスタンスの VPC が CEN インスタンスにアタッチされていることを確認してください。さらに、データソースとデータデスティネーション間のエンドツーエンドの接続を確保するために、リージョン間の接続の帯域幅が構成されていることを確認してください。

    そうでない場合、CEN インスタンスが自動的に作成され、デスティネーションの Message Queue for Apache Kafka インスタンスと専用リソースグループが存在する Elastic Compute Service (ECS) インスタンスの VPC が CEN インスタンスにアタッチされ、エンドツーエンドの接続が確保されます。自動的に作成された CEN インスタンスの帯域幅は、帯域幅が手動で構成されていないため、非常に低くなります。これにより、MySQL ソースコネクタの作成時またはコネクタの実行時に接続エラーが発生する可能性があります。

  • データソースとデスティネーションの Message Queue for Apache Kafka インスタンスは、同じリージョンにある場合があります。この場合、MySQL ソースコネクタを作成すると、データソースまたは Message Queue for Apache Kafka インスタンスの VPC に ENI が自動的に作成されます。ENI は、専用リソースグループが存在する ECS インスタンスにも自動的にバインドされます。これにより、データソースとデータデスティネーション間のエンドツーエンドの接続が確保されます。
• DataWorks の専用リソースグループ
  • DataWorks では、各専用リソースグループを使用して最大 3 つの MySQL ソースコネクタを実行できます。MySQL ソースコネクタを作成するときに、既存のリソースグループで 3 つ未満の MySQL ソースコネクタが実行されていることが DataWorks によって検出されると、DataWorks はこのリソースグループを使用して新しく作成された MySQL ソースコネクタを実行します。
  • DataWorks の各専用リソースグループは、最大 2 つの VPC の ENI に関連付けることができます。リソースグループに関連付ける ENI と、リソースグループに既に関連付けられている ENI の CIDR ブロックが重複している場合、またはその他の技術的な制限により、DataWorks はリソースグループを使用して MySQL ソースコネクタを作成できない場合があります。この場合、このリソースグループで 3 つ未満のコネクタが実行されている場合でも、DataWorks は MySQL ソースコネクタを作成できるようにリソースグループを作成します。

MySQL ソースコネクタを作成してデプロイする

1. ApsaraMQ for Kafka コンソール にログインします。

2. リソースの分布概要 ページの セクションで、管理する ApsaraMQ for Kafka インスタンスが存在するリージョンを選択します。

3. 左側のナビゲーションペインで、Connector タスクリスト をクリックします。

4. Connector タスクリスト ページで、インスタンスの選択 ドロップダウンリストからコネクタが属するインスタンスを選択し、Connector の作成 をクリックします。

5. Connector の作成 ウィザードで、次の操作を実行します。
   1. 基本情報の設定 ステップで、[名前] フィールドにコネクタ名を入力し、次へ をクリックします。

      パラメーター	説明	例
      名前	コネクタの名前。次の命名規則に基づいてコネクタ名を指定します。 コネクタ名は 1 ～ 48 文字で、数字、小文字、ハイフン (-) を使用できます。名前はハイフン (-) で始めることはできません。 各コネクタ名は、ApsaraMQ for Kafka インスタンス内で一意である必要があります。 コネクタは、connect-コネクタ名 形式で名前が付けられたコンシューマーグループを使用する必要があります。グループ このコンシューマーグループを作成していない場合、Message Queue for Apache Kafka は自動的にコンシューマーグループを作成します。グループ	kafka-source-mysql
      インスタンス	Message Queue for Apache Kafka インスタンスに関する情報。デフォルトでは、インスタンスの名前と ID が表示されます。	demo alikafka_post-cn-st21p8vj****

   2. ソースサービスの設定 ステップで、ソースサービスとして [ApsaraDB RDS for MySQL] を選択し、次の表に示すパラメーターを構成して、Next をクリックします。

      パラメーター	説明	例
      RDS インスタンスが所属するリージョン	ApsaraDB RDS for MySQL インスタンスのリージョン ID。ドロップダウンリストからリージョン ID を選択します。	中国 (深セン)
      ApsaraDB RDS のインスタンス ID	データを同期する ApsaraDB RDS for MySQL インスタンスの ID。	rm-wz91w3vk6owmz****
      データベース名	データを同期する ApsaraDB RDS for MySQL データベースの名前。	mysql-to-kafka
      データベースアカウント	ApsaraDB RDS for MySQL データベースへのアクセスに使用するアカウントのユーザー名。	mysql_to_kafka
      データベースアカウントのパスワード	ApsaraDB RDS for MySQL データベースへのアクセスに使用するアカウントのパスワード。	該当なし
      データベーステーブル	データを同期する 1 つ以上の ApsaraDB RDS for MySQL テーブルの名前。複数のテーブル名はカンマ (,) で区切ります。 ソーステーブルとデスティネーショントピックは、1 対 1 のマッピングに基づいて関連付けられます。	mysql_tbl
      データテーブルの自動追加	システムがデータベース内の他のテーブルを自動的に追加できる正規表現。システムは、データベースにテーブルが作成され、正規表現と一致することを検出すると、テーブル内のデータを Message Queue for Apache Kafka に自動的に同期します。 .* を使用して、データベース内のすべてのテーブルを指定できます。	.*
      Topic プレフィックス	ApsaraMQ for Kafka でコネクタ用に自動的に作成されるトピックに名前を付けるために使用されるプレフィックス。各トピック名は、プレフィックスと ApsaraDB RDS for MySQL データベース内の対応するソーステーブルの名前で構成されます。プレフィックスがグローバルに一意であることを確認してください。	mysql

      重要

      ApsaraDB RDS for MySQL データベースアカウントには、少なくとも次の権限が付与されていることを確認してください。

      • SELECT
      • REPLICATION SLAVE
      • REPLICATION CLIENT

      次のステートメントは、アカウントに権限を付与する方法の例を示しています。

      GRANT SELECT, REPLICATION SLAVE, REPLICATION CLIENT ON *.* TO 'アカウントのユーザー名' @'%'; // データベースアカウントに SELECT、REPLICATION SLAVE、REPLICATION CLIENT の権限を付与します。

   3. ターゲットサービスの設定 ステップで、データを同期するデスティネーションの ApsaraMQ for Kafka インスタンスが表示されます。情報を確認し、作成 をクリックします。
6. Connector タスクリスト ページに移動し、作成したコネクタを見つけて、デプロイ操作 列の をクリックします。
   Connector タスクリスト ページで、コネクタの ステータス パラメーターの値が 実行中 として表示されている場合、コネクタは作成されています。

   説明 コネクタの作成に失敗した場合は、このトピックで説明されているすべての前提条件が満たされているかどうかを確認してください。

   MySQL ソースコネクタの構成を変更する必要がある場合は、操作 列の タスクの設定 をクリックします。リダイレクトされた DataWorks コンソールで、コネクタの構成を変更します。

結果の確認

1. ApsaraDB RDS for MySQL データベースのソーステーブルにデータを挿入します。
   次のサンプルコードは、ソーステーブルにデータを挿入する方法の例を示しています。

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

   SQL ステートメントの詳細については、MySQL でよく使用される SQL ステートメント をご参照ください。
2. ApsaraMQ for Kafka のメッセージクエリ機能を使用して、ApsaraDB RDS for MySQL データベースのソーステーブルのデータが ApsaraMQ for Kafka インスタンスの対応するトピックに同期されているかどうかを確認します。
   詳細については、メッセージのクエリ をご参照ください。
   次のサンプルコードは、ApsaraDB RDS for MySQL のソーステーブルから ApsaraMQ for Kafka のトピックに同期されたデータの例を示しています。メッセージの構造とフィールドの詳細については、付録: メッセージ形式 をご参照ください。

   {
       "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"
   }