Flink CDC データインジェストジョブのクイックスタート - Realtime Compute for Apache Flink

前提条件

Flink ワークスペースが作成されていること。詳細については、「Realtime Compute for Apache Flink の有効化」をご参照ください。
ソースデータストアとシンクデータストア
- ApsaraDB RDS for MySQL インスタンスが作成されていること。詳細については、「ApsaraDB RDS for MySQL インスタンスのクイック作成」をご参照ください。
- EMR Serverless StarRocks インスタンスが作成されていること。詳細については、「手順」をご参照ください。
説明
ApsaraDB RDS for MySQL インスタンスと EMR Serverless StarRocks インスタンスは、お使いの Flink ワークスペースと同じ Virtual Private Cloud (VPC) 内にある必要があります。異なる VPC にある場合は、ネットワーク接続を確立し、ApsaraDB RDS for MySQL インスタンスの許可リストを設定する必要があります。詳細については、「VPC をまたいで他のサービスにアクセスする方法」、「インターネットへのアクセス方法」、および「許可リストの設定方法」をご参照ください。

背景情報

お使いの ApsaraDB RDS for MySQL インスタンスに、order_dw_mysql という名前のデータベースがあり、orders、orders_pay、product_catalog の 3 つの業務テーブルが含まれていると仮定します。これらのテーブルとそのデータを StarRocks の order_dw_sr データベースに同期するには、次の手順に従ってください。

ステップ1：ApsaraDB RDS for MySQL でのテストデータの準備
ステップ2：Flink CDC データインジェストジョブの開発
ステップ3：Flink CDC データインジェストジョブの開始
ステップ4：StarRocks での同期結果の検証

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

データベースとアカウントの作成

order_dw_mysql という名前のデータベースと、それに対する読み取り/書き込み権限を持つ標準アカウントを作成します。詳細については、「データベースとアカウントの作成」および「データベースの管理」をご参照ください。
Data Management (DMS) を使用して ApsaraDB RDS for MySQL インスタンスにログインします。

詳細については、「DMS を使用した ApsaraDB RDS for MySQL インスタンスへのログイン」をご参照ください。

SQL Console ウィンドウで、次のコマンドを入力し、[Execute] をクリックして 3 つの業務テーブルを作成し、データを挿入します。

CREATE TABLE `orders` (
  order_id bigint not null primary key,
  user_id varchar(50) not null,
  shop_id bigint not null,
  product_id bigint not null,
  buy_fee numeric(20,2) not null,   
  create_time timestamp not null,
  update_time timestamp not null default now(),
  state int not null 
);
CREATE TABLE `orders_pay` (
  pay_id bigint not null primary key,
  order_id bigint not null,
  pay_platform int not null, 
  create_time timestamp not null
);
CREATE TABLE `product_catalog` (
  product_id bigint not null primary key,
  catalog_name varchar(50) not null
);
-- データを準備します
INSERT INTO product_catalog VALUES(1, 'phone_aaa'),(2, 'phone_bbb'),(3, 'phone_ccc'),(4, 'phone_ddd'),(5, 'phone_eee');
INSERT INTO orders VALUES
(100001, 'user_001', 12345, 1, 5000.05, '2023-02-15 16:40:56', '2023-02-15 18:42:56', 1),
(100002, 'user_002', 12346, 2, 4000.04, '2023-02-15 15:40:56', '2023-02-15 18:42:56', 1),
(100003, 'user_003', 12347, 3, 3000.03, '2023-02-15 14:40:56', '2023-02-15 18:42:56', 1),
(100004, 'user_001', 12347, 4, 2000.02, '2023-02-15 13:40:56', '2023-02-15 18:42:56', 1),
(100005, 'user_002', 12348, 5, 1000.01, '2023-02-15 12:40:56', '2023-02-15 18:42:56', 1),
(100006, 'user_001', 12348, 1, 1000.01, '2023-02-15 11:40:56', '2023-02-15 18:42:56', 1),
(100007, 'user_003', 12347, 4, 2000.02, '2023-02-15 10:40:56', '2023-02-15 18:42:56', 1);
INSERT INTO orders_pay VALUES
(2001, 100001, 1, '2023-02-15 17:40:56'),
(2002, 100002, 1, '2023-02-15 17:40:56'),
(2003, 100003, 0, '2023-02-15 17:40:56'),
(2004, 100004, 0, '2023-02-15 17:40:56'),
(2005, 100005, 0, '2023-02-15 18:40:56'),
(2006, 100006, 0, '2023-02-15 18:40:56'),
(2007, 100007, 0, '2023-02-15 18:40:56');

ステップ2：Flink CDC ジョブの開発

Realtime Compute for Apache Flink 管理コンソールにログインします。
[Console] をクリックしてプロジェクトのワークスペースに移動します。
ナビゲーションペインで、[Development] > [Data Ingestion] を選択します。
アイコンをクリックし、[New Draft with Template] をクリックし、[MySQL to StarRocks data synchronization] を選択してから [Next] をクリックします。
[Job Name] と [Storage Location] を入力し、[Engine Version] を選択してから [OK] をクリックします。

YAML ジョブコードを設定します。

次のコードは、MySQL の order_dw_mysql データベースから StarRocks の order_dw_sr データベースにすべてのテーブルを同期する例を示しています。

source:
  type: mysql
  hostname: rm-bp1rk934iidc3****.mysql.rds.aliyuncs.com
  port: 3306
  username: ${secret_values.mysqlusername}
  password: ${secret_values.mysqlpassword}
  tables: order_dw_mysql.\.*
  server-id: 8601-8604
  # (任意) 増分フェーズ中に新しく作成されたテーブルからデータを同期します。
  scan.binlog.newly-added-table.enabled: true
  # (任意) テーブルと列のコメントを同期します。
  include-comments.enabled: true
  # (任意) TaskManager の OutOfMemory エラーを防ぐために、unbounded チャンクの配布を優先します。
  scan.incremental.snapshot.unbounded-chunk-first.enabled: true
  # (任意) キャプチャ対象外テーブルのチェンジログの逆シリアル化をスキップします。
  scan.only.deserialize.captured.tables.changelog.enabled: true 
sink:
  type: starrocks
  name: StarRocks Sink
  jdbc-url: jdbc:mysql://fe-c-b76b6aa51807****-internal.starrocks.aliyuncs.com:9030
  load-url: fe-c-b76b6aa51807****-internal.starrocks.aliyuncs.com:8030
  username: ${secret_values.starrocksusername}
  password: ${secret_values.starrockspassword}
  table.create.properties.replication_num: 1
  sink.buffer-flush.interval-ms: 5000 # 5 秒ごとにデータをフラッシュします。
route:
  - source-table: order_dw_mysql.\.*
    sink-table: order_dw_sr.<>
    replace-symbol: <>
    description: source_db 内のすべてのテーブルを sink_db にルーティングします
pipeline:
  name: Sync MySQL Database to StarRocks

次の表に、この例で必要な設定パラメーターを示します。データインジェストパラメーターの詳細については、「MySQL」および「StarRocks」をご参照ください。

説明

YAML ジョブはプロジェクト変数のみをサポートします。変数を使用すると、パスワードなどの情報がプレーンテキストで表示されるのを防ぐことができます。詳細については、「変数管理」をご参照ください。

カテゴリ	パラメーター	説明	値の例
ソース	hostname	MySQL データベースの IP アドレスまたはホスト名。内部エンドポイントの使用を推奨します。	`rm-bp1rk934iidc3****.mysql.rds.aliyuncs.com`
	port	MySQL データベースサービスのポート番号。	3306
	username	MySQL データベースのユーザー名とパスワード。ステップ1：MySQL テストデータの準備で作成したアカウントの認証情報を使用します。	`${secret_values.mysqlusername}`
	password		`${secret_values.mysqlpassword}`
	tables	MySQL テーブルの名前。正規表現を使用して、複数のテーブルからデータを読み取ることができます。このトピックでは、order_dw_mysql データベース内のすべてのテーブルとデータが同期されます。	order_dw_mysql.\.*
	server-id	データベースクライアント接続用の一意の数値 ID。	8601-8604
シンク	jdbc-url	JDBC 接続 URL。フロントエンド (FE) の IP アドレスとクエリポートを `jdbc:mysql://ip:port` の形式で指定します。 E-MapReduce コンソールの [Instance Details] タブで、ターゲットインスタンスの FE の[内部エンドポイント]と[クエリポート]を表示できます。	`jdbc:mysql://fe-c-b76b6aa51807****-internal.starrocks.aliyuncs.com:9030`
	load-url	FE ノードへの接続に使用される HTTP サービス URL。 E-MapReduce コンソールの [Instance Details] タブで、ターゲットインスタンスの FE の[内部エンドポイント]と[HTTP ポート]を表示できます。	`fe-c-b76b6aa51807****-internal.starrocks.aliyuncs.com:8030`
	username	StarRocks への接続に使用する認証情報。 StarRocks インスタンスの作成時に設定した認証情報を使用します。説明この例では、変数を使用して、認証情報がプレーンテキストで表示されるのを防ぎます。詳細については、「変数管理」をご参照ください。	`${secret_values.starrocksusername}`
	password		`${secret_values.starrockspassword}`
	sink.buffer-flush.interval-ms	内部バッファーのフラッシュ間隔。この例ではデータ量が少ないため、短い間隔 (5 秒) を使用して、結果をすばやく確認できるようにしています。	5000
ルート	source-table	ルートするソーステーブル。正規表現を使用して複数のテーブルを照合できます。たとえば、`order_dw_mysql.\.*` は `order_dw_mysql` データベース内のすべてのテーブルをルートします。	order_dw_mysql.\.*
	sink-table	ルーティングされたデータの宛先テーブルパターン。 `replace-symbol` パラメーターで定義されたシンボルを各ソーステーブル名のプレースホルダーとして使用して、多対多のルーティングを実現できます。ルーティングルールの詳細については、「ルートモジュール」をご参照ください。	order_dw_sr.<>
	replace-symbol	パターンマッチングで使用されるソーステーブル名のプレースホルダー。	<>

[Deploy] をクリックします。

ステップ3：Flink CDC ジョブの開始

[Data Ingestion] ページで [Deploy] をクリックし、表示されるダイアログボックスで [OK] をクリックします。
[O&M] > [Deployments] ページで、対象の YAML ジョブを見つけ、[Actions] 列の [Start] をクリックします。
[Start] をクリックします。

この例では、[Initial Mode] を選択します。パラメーターの詳細については、「ジョブの開始」をご参照ください。ジョブが開始されたら、[Deployments] ページでそのステータスを監視できます。

ステップ4：StarRocks での結果の検証

ジョブが [RUNNING] 状態になった後、StarRocks でデータを確認できます。

EMR StarRocks Manager を使用した EMR Serverless StarRocks インスタンスへの接続。
ナビゲーションペインで [SQL Editor] をクリックします。[Database] タブで、更新アイコンをクリックします。

default_catalog の下に order_dw_sr という名前のデータベースが表示されます。

[Query List] タブで [+ File] をクリックして [Query Script] を作成します。次の SQL 文を入力して [Run] をクリックします。

SELECT * FROM default_catalog.order_dw_sr.orders order by order_id;
SELECT * FROM default_catalog.order_dw_sr.orders_pay order by pay_id;
SELECT * FROM default_catalog.order_dw_sr.product_catalog order by product_id;

コマンドの下に表示される結果を確認します。

この結果から、MySQL データベースのテーブルとデータが StarRocks に存在することがわかります。

同期されたテーブルには、default_catalog.order_dw_sr.orders、default_catalog.order_dw_sr.orders_pay、default_catalog.order_dw_sr.product_catalog が含まれます。SELECT 文を実行して各テーブルをクエリし、データ整合性を検証できます。

Realtime Compute for Apache Flink:Flink CDC データインジェストジョブ