Data Lake Formation:Flink と DLF を使用したデータレイクへのデータインポートとデータ分析

ソーステーブルと結果テーブルの作成

1. 左側のナビゲーションウィンドウで、[開発] > [スクリプト] をクリックします。

2. SQL 編集エリアで、次の SQL を入力し、[実行] をクリックします。

   -- MySQL から変更イベントを読み取るソーステーブルを作成します
   CREATE TABLE IF NOT EXISTS student_source (
     id INT,
     name VARCHAR(256),
     age INT,
     PRIMARY KEY (id) NOT ENFORCED
   )
   WITH (
     'connector' = 'mysql-cdc',
     -- ご利用の ApsaraDB RDS for MySQL インスタンスのエンドポイントに置き換えてください
     'hostname' = 'rm-xxxxxxxx.mysql.rds.aliyuncs.com',
     'port' = '3306',
     'username' = '<RDS username>',
     'password' = '<RDS password>',
     'database-name' = '<RDS database>',
     -- ステップ 1 で作成したソーステーブルの名前に設定してください
     'table-name' = 'student'
   );
   
   -- DLF カタログにターゲットデータベースを作成します
   -- DLF カタログ名が異なる場合は、'dlf' を実際のカタログ名に置き換えてください
   CREATE DATABASE IF NOT EXISTS dlf.dlf_testdb;
   
   -- DLF カタログに Apache Hudi の結果テーブルを作成します
   CREATE TABLE IF NOT EXISTS dlf.dlf_testdb.student_hudi (
     id    BIGINT PRIMARY KEY NOT ENFORCED,
     name  STRING,
     age   BIGINT
   ) WITH (
     'connector' = 'hudi'
   );

   プレースホルダー	説明	例
   rm-xxxxxxxx.mysql.rds.aliyuncs.com	ApsaraDB RDS for MySQL インスタンスのエンドポイント	rm-bp1xxxxxxxx.mysql.rds.aliyuncs.com
   <RDS username>	ソーステーブルへの読み取りアクセス権を持つ MySQL ユーザー	admin
   <RDS password>	MySQL ユーザーのパスワード	—
   <RDS database>	ソースデータベースの名前	testdb

   テーブルが作成されると、両方とも [Catalogs] の下に表示されます。

   

   mysql-cdc コネクタパラメーターの完全なリストについては、「MySQL ソースコネクタ」をご参照ください。Hudi の結果テーブルのパラメーターについては、「Hudi コネクタ (廃止予定)」をご参照ください。

ストリーミングジョブの作成とデプロイ

1. 左側のナビゲーションウィンドウで、[開発] > [ETL] をクリックします。

2. [新規] をクリックし、[新規ドラフト] ダイアログボックスで [空白のストリームドラフト] を選択して、[次へ] をクリックします。

3. ドラフト構成を入力し、[作成] をクリックします。

4. SQL 編集エリアで、次の INSERT 文を入力します。

   -- MySQL ソースから Hudi の結果テーブルに変更をストリーミングします
   INSERT INTO dlf.dlf_testdb.student_hudi
   SELECT * FROM student_source /*+ OPTIONS('server-id'='123456') */;

5. SQL 編集エリアの右上隅にある [デプロイ] をクリックします。[ドラフトのデプロイ] ダイアログボックスで、必須フィールドを入力し、[確認] をクリックします。

ジョブの開始

1. 左側のナビゲーションウィンドウで、[運用保守] > [デプロイメント] をクリックします。

2. 対象のジョブの [アクション] 列で、[開始] をクリックします。

3. [初期モード] を選択し、[開始] をクリックします。

ジョブのステータスが [実行中] に変わると、Flink は MySQL からの変更をアクティブにキャプチャし、Hudi の結果テーブルに書き込みます。起動パラメーターの詳細については、「デプロイメントの開始」をご参照ください。