E-MapReduce:Use the CTAS and CDAS statements of Realtime Compute for Apache Flink to synchronize data from an ApsaraDB RDS for MySQL instance to a StarRocks cluster

スキーマ変更の動作

CTAS は、スキーマ変更を伝播するための固定ポリシーを使用します。次の表は、自動的に伝播されるものとされないものをまとめたものです。

サポートされているスキーマ変更

サポートされていないスキーマ変更

ソーステーブルで次のいずれかの変更が発生した場合、StarRocks の送信先テーブルを削除し、CTAS ジョブを再起動します。Flink はテーブルを再作成し、すべての既存データを再同期します。

• 列のデータの型を変更する (例: VARCHAR から BIGINT、または NOT NULL から NULLABLE)

• プライマリキーやインデックスなどの制約を変更する

• NULL 不可列を追加または削除する

• DDL (データ定義言語) ステートメントでフィールド長を調整する

CTAS は、連続するデータレコードのスキーマを比較することでスキーマ変更を検出します。DDL ステートメントのタイプは解析しません。結果として、次のようになります。

列が削除され、その間にデータ変更なしで再追加された場合、CTAS はこれをスキーマ変更なしとして扱います。

スキーマ変更は、変更後に新しいデータがソーステーブルに到着した場合にのみ伝播されます。

フィールドタイプのマッピングについては、「Apache Flink からデータを継続的にロードする」をご参照ください。

CTAS が複数の MySQL テーブルをマージする場合、Flink はソースを追跡するために、送信先テーブルに _db_name および _table_name 列を自動的に先頭に追加します。3 番目の列から独自の列順序を定義します。

At-least-once セマンティクス (低レイテンシーシナリオに推奨)

データは構成可能なフラッシュ間隔で書き込まれます。メモリ使用量は少ないですが、障害発生時に重複書き込みが発生する可能性があります。

/* 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'
) */;

1 回限りのセマンティクス (データが重要なシナリオに推奨)

障害発生時にデータ損失や重複はありません。データの可視性はチェックポイント間隔に依存します。

/* 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'
) */;

チェックポイントの構成オプションについては、「チェックポイント」をご参照ください。

シンプルモード (クイックセットアップに推奨)

Flink は MySQL スキーマからテーブル定義を推論します。エンジン、キー、またはディストリビューションを手動で指定する必要はありません。シンプルモードではパーティションテーブルはサポートされていません。代わりに通常モードを使用してパーティションを作成してください。

/* 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'      = 'test',
  'password'      = '123456',
  'database-name' = 'test_cdc',
  'table-name'    = 'runoob_tbl'
) */;

WITH 句パラメーター

お使いの Flink バージョンが vvr-6.0.5-flink-1.15 より前の場合、WITH 句に 'sink.use.new-apiapi' = 'false' を追加します。追加の sink パラメーターについては、「Apache Flink からデータを継続的にロードする」をご参照ください。

OPTIONS 句パラメーター

OPTIONS 句は MySQL CDC ソースを構成します。

デプロイメントの公開と開始

1. [詳細設定] タブの [下書きエディタ] ページで、[エンジンバージョン] を vvr-6.0.3-flink-1.15 以降に設定します。

2. 「下書きエディタ」ページの右上隅で、[公開] をクリックします。

3. [デプロイメント] ページで、新しいデプロイメントを見つけ、[操作] 列の [開始] をクリックします。

初期データの検証

StarRocks テーブルをクエリして、シード行が同期されたことを確認します。

SELECT * FROM runoob_tbl1;

期待される出力:

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

INSERT の検証

ApsaraDB RDS for MySQL インスタンスの [SQL Console] タブで、行を挿入します：

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

両方の行が表示されることを確認するために StarRocks をクエリします。

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 |
+-----------+--------------+---------------+-----------------+---------+

UPDATE の検証

MySQL SQL コンソールで、行を更新します。

UPDATE runoob_tbl SET runoob_title = 'new' WHERE runoob_id = 18;

変更を確認するために StarRocks をクエリします。

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 |
+-----------+--------------+---------------+-----------------+---------+

DELETE の検証

MySQL SQL コンソールで、行を削除します。

DELETE FROM runoob_tbl WHERE runoob_id = 1;

行が削除されたことを確認するために StarRocks をクエリします。

SELECT * FROM runoob_tbl1;

期待される出力:

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

スキーマ変更の検証: NULL 許容列の追加

MySQL SQL コンソールで、NULL 許容列を追加し、その列に値を持つ行を挿入します。

ALTER TABLE `runoob_tbl` ADD COLUMN `add_col2` INT;

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);

新しい列が表示され、既存の行が NULL を示すことを確認するために StarRocks をクエリします。

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 |
+-----------+--------------+---------------+-----------------+---------+----------+

add_col2 列は、それを含む最初のデータ行が到着したときに StarRocks に自動的に追加されました。