E-MapReduce:CTAS および CDAS を使用して MySQL から StarRocks へデータを同期する

前提条件

• Realtime Compute for Apache Flink（フルマネージド）を有効化し、Flink クラスターを作成済みである必要があります。詳細については、「Realtime Compute for Apache Flink（フルマネージド）の有効化」および「Flink SQL デプロイメントのクイックスタート」をご参照ください。

• StarRocks クラスターを作成しました。詳細については、「StarRocks クラスターの作成」をご参照ください。

• ApsaraDB RDS for MySQL インスタンスを作成済みである必要があります。詳細については、「ApsaraDB RDS for MySQL インスタンスの作成」をご参照ください。

制限事項

• Flink クラスター、StarRocks クラスター、および ApsaraDB RDS for MySQL インスタンスは、同じ VPC 内に配置されている必要があります。

• ApsaraDB RDS for MySQL インスタンスのバージョンは 5.7 以降である必要があります。

• StarRocks クラスターはインターネットアクセスを有効にする必要があります。

• Flink クラスターの Flink バージョンは 1.15-vvr-6.0.3 以降である必要があります。

ステップ 1：テストデータの準備

1. テスト用データベースおよびアカウントを作成します。詳細については、「ApsaraDB RDS for MySQL インスタンスのデータベースおよびアカウントの作成」をご参照ください。

   データベースおよびアカウントを作成後、テストアカウントに読み取りおよび書き込み権限を付与します。

   説明

   本トピックでは、データベース名を test_cdc、アカウント名を test とします。

2. テストアカウントを使用して MySQL インスタンスに接続します。詳細については、「DMS を使用した ApsaraDB RDS for MySQL インスタンスへのログイン」をご参照ください。

3. MySQL で以下のコマンドを実行してデータテーブルを作成します。

   use test_cdc;
   
   CREATE TABLE IF NOT EXISTS `runoob_tbl`(
      `runoob_id` INT UNSIGNED AUTO_INCREMENT,
      `runoob_title` VARCHAR(100) NOT NULL,
      `runoob_author` VARCHAR(40) NOT NULL,
      `submission_date` DATE,
      `add_col` int DEFAULT NULL,
      PRIMARY KEY ( `runoob_id` )
   )ENGINE=InnoDB DEFAULT CHARSET=utf8;
   
   
   INSERT INTO test_cdc.`runoob_tbl` (`runoob_id`,`runoob_title`,`runoob_author`,`submission_date`,`add_col`) values (18,'first','tom','2022-06-22 17:13:44',3)

4. SSH を使用して StarRocks クラスターにログインします。詳細については、「クラスターへのログイン」をご参照ください。

5. 以下のコマンドを実行して StarRocks クラスターに接続します。

   mysql -h127.0.0.1 -P 9030 -uroot

6. 以下のコマンドを実行してユーザーを作成し、権限を付与します。

   CREATE DATABASE test_cdc;
   CREATE USER 'test' IDENTIFIED by '123456';
   GRANT CREATE TABLE ON DATABASE test_cdc TO test;

ステップ 2：カタログの作成

• MySQL カタログ

  • 例

    CREATE CATALOG mysql WITH (
      'type' = 'mysql',
      'hostname' = 'rm-2zepd6e20u3od****.mysql.rds.aliyuncs.com',
      'port' = '3306',
      'username' = 'emr-test',
      'password' = '123456',
      'default-database' = 'test_cdc'
    );

  • パラメーター

    パラメーター	説明
    type	カタログタイプ。値を mysql に設定します。
    hostname	ApsaraDB RDS for MySQL インスタンスの内部エンドポイント。ApsaraDB RDS コンソールの [データベース接続] ページから内部エンドポイントをコピーできます。例：rm-2zepd6e20u3od****.mysql.rds.aliyuncs.com。
    port	MySQL データベースサービスのポート番号。デフォルト値は 3306 です。
    username	MySQL データベースサービスへのアクセスに使用するユーザー名。 ステップ 1：テストデータの準備 で作成したアカウントのユーザー名を使用します。この例では test を使用します。
    password	MySQL データベースサービスへのアクセスに使用するパスワード。 ステップ 1：テストデータの準備 で作成したアカウントのパスワードを使用します。
    default-database	デフォルトの MySQL データベース名。 ステップ 1：テストデータの準備 で作成したデータベース名を使用します。この例では test_cdc を使用します。

• StarRocks カタログ

  • 例

    CREATE CATALOG sr  WITH (
      'type' = 'starrocks',
      'endpoint' = '172.16.**.**:9030',
      'username' = 'test',
      'password' = '123456',
      'dbname' = 'test_cdc'
    );

  • パラメーター

    パラメーター	説明
    type	カタログタイプ。値を starrocks に設定します。
    endpoint	StarRocks フロントエンド（FE）の IP アドレスおよびポート。
    username	StarRocks クラスターへのアクセスに使用するユーザー名。 ステップ 1：テストデータの準備 で作成したアカウントのユーザー名を使用します。この例では test を使用します。
    password	StarRocks データベースサービスのパスワード。 ステップ 1：テストデータの準備 で作成したアカウントのパスワードを使用します。
    dbname	StarRocks データベース名。 ステップ 1：テストデータの準備 で作成したデータベース名を使用します。この例では test_cdc を使用します。

ステップ 3：デプロイメントの作成および公開

1. Realtime Compute for Apache Flink コンソールの下書きエディター画面で、CTAS 文を記述します。

   以下に CTAS 文の例を 3 つ示します。

   • at-least-once セマンティクス：sink.buffer-flush.interval-ms オプションを使用して、StarRocks へのデータ書き込み間隔を設定します。このオプションにより、遅延およびメモリ使用量を削減できます。

     /*
           At-least-once semantics
     */
     
     use CATALOG sr;
     
     CREATE TABLE IF NOT EXISTS runoob_tbl_sr with (
     'starrocks.create.table.properties'=' engine = olap primary key(runoob_id) distributed by hash(runoob_id ) buckets 8',
     'database-name'='test_cdc',
     'jdbc-url'='jdbc:mysql://172.16.**.**:9030',
     'load-url'='172.16.**.**:18030',
     'table-name'='runoob_tbl_sr',
     'username'='test',
     'password' = '123456',
     'sink.buffer-flush.interval-ms' = '5000',
     'sink.properties.row_delimiter' = '\x02',
     'sink.properties.column_separator' = '\x01'
     )
      as table mysql.test_cdc.runoob_tbl  /*+ OPTIONS (   'connector' = 'mysql-cdc',
       'hostname' = 'rm-2zepd6e20u3od****.mysql.rds.aliyuncs.com',
       'port' = '3306',
       'username' = 'test',
       'password' = '123456',
       'database-name' = 'test_cdc',
       'table-name' = 'runoob_tbl'  )*/;
                                     

   • exactly-once セマンティクス：チェックポイント間隔を設定する必要があります。これにより障害発生時のデータ損失および重複を防止できますが、データの可視性はチェックポイント間隔に依存します。詳細については、「Checkpointing」をご参照ください。

     /*
           Exactly-once semantics.
     */
     set 'execution.checkpointing.interval' = '1 min';
     set 'execution.checkpointing.mode' = 'EXACTLY_ONCE';
     set 'execution.checkpointing.timeout' = '10 min';
     
     use CATALOG sr;
     
     CREATE TABLE IF NOT EXISTS runoob_tbl with (
     'starrocks.create.table.properties'=' engine = olap primary key(runoob_id) distributed by hash(runoob_id ) buckets 8',
     'database-name'='test_cdc',
     'jdbc-url'='jdbc:mysql://172.16.**.**:9030',
     'load-url'='172.16.**.**:18030',
     'table-name'='runoob_tbl',
     'username'='test',
     'password' = '123456',
     'sink.semantic' = 'exactly-once',
     'sink.properties.row_delimiter' = '\x02',
     'sink.properties.column_separator' = '\x01'
     )
      as table mysql.test_cdc.runoob_tbl  /*+ OPTIONS (   'connector' = 'mysql-cdc',
       'hostname' = 'rm-2zepd6e20u3od****.mysql.rds.aliyuncs.com',
       'port' = '3306',
       'username' = 'test',
       'password' = '123456',
       'database-name' = 'test_cdc',
       'table-name' = 'runoob_tbl'  )*/;
                                     

   • シンプルモード：テーブル作成時にフィールドを定義する必要はありません。テーブルスキーマは MySQL からコピーされます。ただし、パーティションは作成できません。パーティションを使用する場合は、通常モードを使用する必要があります。

     /*
           The two preceding examples use normal mode. This example demonstrates simple mode.
     */
     
     use CATALOG sr;
     
     CREATE TABLE IF NOT EXISTS runoob_tbl1 with (
     'starrocks.create.table.properties'='buckets 8',
     'starrocks.create.table.mode'='simple',
      'database-name'='test_cdc',
     'jdbc-url'='jdbc:mysql://172.16.**.**:9030',
     'load-url'='172.16.**.**:18030',
     'table-name'='runoob_tbl_sr',
     'username'='test',
     'password' = '123456',
     'sink.buffer-flush.interval-ms' = '5000',
     'sink.properties.row_delimiter' = '\x02',
     'sink.properties.column_separator' = '\x01'
     )
      as table mysql.test_cdc.runoob_tbl  /*+ OPTIONS (   'connector' = 'mysql-cdc',
       'hostname' = 'rm-2zepd6e20u3od****.mysql.rds.aliyuncs.com',
       'port' = '3306',
       'username' = 'emr-test',
       'password' = '123456',
       'database-name' = 'test_cdc',
       'table-name' = 'runoob_tbl'  )*/;
                                     

   表 1. WITH パラメーター

   パラメーター	必須	説明
   starrocks.create.table.properties	はい	StarRocks の CREATE TABLE 文におけるカラム定義を除くサフィックス定義。例：engine、key、buckets など。
   database-name	はい	StarRocks データベース名。 この例では test_cdc を使用します。
   jdbc-url	はい	StarRocks でのクエリ操作に使用されます。 例：jdbc:mysql://172.16.**.**:9030。172.16.**.** の部分は StarRocks クラスターの内部 IP アドレスです。
   load-url	はい	StarRocks フロントエンド（FE）の IP アドレスおよび HTTP ポート。形式は StarRocks クラスターの内部 IP アドレス:ポート です。本トピックではポート 8030 を例として使用します。クラスターバージョンに応じてポートを選択してください。 18030：EMR V5.9.0 以降および EMR V3.43.0 以降。 8030：EMR V5.8.0 以前および EMR V3.42.0 以前。 説明 ポートの詳細については、「UI およびポート」をご参照ください。
   sink.semantic	いいえ	データ整合性セマンティクスを保証するには、exactly-once に設定します。デフォルトは at-least-once です。
   starrocks.create.table.mode	いいえ	サポートされる値： normal（デフォルト）：starrocks.create.table.properties オプションに engine、key、buckets などの完全な構成を指定する必要があります。 simple：エンジンはデフォルトで olap、キーの種類はデフォルトで primary key に設定されます。プライマリキーは MySQL テーブルから継承されます。デフォルトでは、すべてのプライマリキー列のハッシュによってテーブルが分散され、パーティションは作成されません。starrocks.create.table.properties オプションで buckets を指定する必要があります。properties などの他の構成はオプションです。

   説明

   • sink.use.new-api パラメーターは、Flink バージョン 1.15-vvr-6.0.5 以降で削除されています。1.15-vvr-6.0.5 より前のバージョンを使用している場合は、WITH パラメーターに 'sink.use.new-api'='false', を追加する必要があります。

   • 他の構成についての詳細については、「Apache Flink からデータを継続的にロードする」をご参照ください。

   表 2. OPTIONS パラメーター

   パラメーター	説明
   connector	コネクタタイプ。値を mysql-cdc に設定します。
   hostname	ApsaraDB RDS for MySQL インスタンスの内部エンドポイント。 ApsaraDB RDS コンソールのインスタンスの [データベース接続] ページから内部エンドポイントをコピーできます。例：rm-bp1nu0c46fn9k****.mysql.rds.aliyuncs.com。
   port	MySQL データベースサービスのポート番号。デフォルト値は 3306 です。
   username	MySQL データベースサービスへのアクセスに使用するユーザー名。 ステップ 1：テストデータの準備 で作成したアカウントのユーザー名を使用します。この例では test を使用します。
   password	MySQL データベースサービスへのアクセスに使用するパスワード。 ステップ 1：テストデータの準備 で作成したアカウントのパスワードを使用します。
   table-name	StarRocks のテーブル名。 ステップ 1：テストデータの準備 で作成したテーブル名を使用します。この例では runoob_tbl を使用します。
   database-name	デフォルトの MySQL データベース名。 ステップ 1：テストデータの準備 で作成したデータベース名を使用します。この例では test_cdc を使用します。

2. 下書きエディター画面の [詳細設定] で、Flink バージョン 1.15-vvr-6.0.3 以降を選択します。

3. [online] をクリックします。

4. [デプロイメント] ページで対象のデプロイメントを見つけ、[操作] 列の [START] をクリックします。

ステップ 4：シナリオデモンストレーション

データのクエリ

1. SSH を使用して StarRocks クラスターにログインします。詳細については、「クラスターへのログイン」をご参照ください。

2. 以下のコマンドを実行して StarRocks クラスターに接続します。

   mysql -h127.0.0.1 -P 9030 -uroot

3. StarRocks CLI で以下のコマンドを実行してテーブルデータを表示します。

   use test_cdc;
   select * from runoob_tbl1;

   出力結果から、MySQL テーブルのデータが StarRocks に同期されていることが確認できます。

   +-----------+--------------+---------------+-----------------+---------+
   | runoob_id | runoob_title | runoob_author | submission_date | add_col |
   +-----------+--------------+---------------+-----------------+---------+
   |        18 | first        | tom           | 2022-06-22      |       3 |
   +-----------+--------------+---------------+-----------------+---------+

挿入されたデータのクエリ

1. ApsaraDB RDS for MySQL インスタンスの SQL コンソールで以下のコマンドを実行してデータを挿入します。

   INSERT INTO runoob_tbl(`runoob_id`,`runoob_title`,`runoob_author`,`submission_date`,`add_col`)  values(1,'second','tom2','2022-06-23',1)

2. StarRocks CLI で以下のコマンドを実行してテーブルデータを表示します。

   select * from runoob_tbl1;

   出力結果から、データが正常に挿入されたことが確認できます。

   +-----------+--------------+---------------+-----------------+---------+
   | runoob_id | runoob_title | runoob_author | submission_date | add_col |
   +-----------+--------------+---------------+-----------------+---------+
   |         1 | second       | tom2          | 2022-06-23      |       1 |
   |        18 | first        | tom           | 2022-06-22      |       3 |
   +-----------+--------------+---------------+-----------------+---------+

データ更新の同期

1. ApsaraDB RDS for MySQL インスタンスの SQL コンソールで以下のコマンドを実行して特定のデータを更新します。

   update runoob_tbl set runoob_title= 'new' where runoob_id = 18

2. StarRocks CLI で以下のコマンドを実行してテーブルデータを表示します。

   select * from runoob_tbl1;

   出力結果から、データ更新が同期されたことが確認できます。

   +-----------+--------------+---------------+-----------------+---------+
   | runoob_id | runoob_title | runoob_author | submission_date | add_col |
   +-----------+--------------+---------------+-----------------+---------+
   |         1 | second       | tom2          | 2022-06-23      |       1 |
   |        18 | new          | tom           | 2022-06-22      |       3 |
   +-----------+--------------+---------------+-----------------+---------+

データ削除の同期

1. ApsaraDB RDS for MySQL インスタンスの SQL コンソールで以下のコマンドを実行して特定のデータを削除します。

   DELETE FROM runoob_tbl WHERE runoob_id = 1

2. StarRocks CLI で以下のコマンドを実行してテーブルデータを表示します。

   select * from runoob_tbl1;

   出力結果から、データ削除が同期されたことが確認できます。

   +-----------+--------------+---------------+-----------------+---------+ 
   | runoob_id | runoob_title | runoob_author | submission_date | add_col | 
   +-----------+--------------+---------------+-----------------+---------+
   |        18 | new          | tom           | 2022-06-22      |       3 | 
   +-----------+--------------+---------------+-----------------+---------+

NULL 許容カラムの追加

1. ApsaraDB RDS for MySQL インスタンスの SQL コンソールで以下のコマンドを実行して NULL 許容カラムを追加します。

   alter table `runoob_tbl` add COLUMN `add_col2` INT;

2. 以下のコマンドを実行してデータを挿入します。

   INSERT INTO runoob_tbl(`runoob_id`,`runoob_title`,`runoob_author`,`submission_date`,`add_col`,`add_col2`)  values(1,'second','tom2','2022-06-23',1,2)

3. StarRocks CLI で以下のコマンドを実行してテーブルデータを表示します。

   select * from runoob_tbl1;

   出力結果から、スキーマ変更が成功したことが確認できます。

   +-----------+--------------+---------------+-----------------+---------+----------+
   | runoob_id | runoob_title | runoob_author | submission_date | add_col | add_col2 | 
   +-----------+--------------+---------------+-----------------+---------+----------+ 
   |        18 | new          | tom           | 2022-06-22      |       3 |     NULL |
   +-----------+--------------+---------------+-----------------+---------+----------+ 
   |         1 | second       | tom2          | 2022-06-23      |       1 |      2   |
   |        18 | first        | tom           | 2022-06-22      |       3 |     NULL |
   +-----------+--------------+---------------+-----------------+---------+----------+ 

CDAS の概要

CDAS 文は CTAS の糖衣構文であり、MySQL データベース全体を単一の Flink デプロイメントとして StarRocks に同期します。including table 構文を使用して、一部のテーブルのみを同期することも可能です。

CREATE DATABASE IF NOT EXISTS sr_db with (
  'starrocks.create.table.properties'=' buckets 8',
  'starrocks.create.table.mode'='simple',
  'jdbc-url'='jdbc:mysql://172.16.**.**:9030',
  'load-url'='172.16.**.**:18030',
  'username'='test',
  'password' = '123456',
  'sink.buffer-flush.interval-ms' = '5000' ,
  'sink.properties.row_delimiter' = '\x02',
  'sink.properties.column_separator' = '\x01'
)
as DATABASE mysql.test_cdc including table  'tabl1','tbl2','tbl3'
/*+ OPTIONS (   'connector' = 'mysql-cdc',
'hostname' = 'rm-2zepd6e20u3od****.mysql.rds.aliyuncs.com',
'port' = '3306',   
'username' = 'test',
'password' = '123456',
'database-name' = 'test_cdc' )*/;