MaxCompute:DLF 1.0 および OSS を使用してデータレイクから Paimon データを読み取る

前提条件

1. OSS サービス が有効になっていること。

2. DLF サービス が有効になっていること。

3. Flink サービス が有効になっていること。

4. MaxCompute プロジェクトが作成済み であり、プロジェクトレベルのメタデータサポートのためのスキーマスイッチが有効になっていること。

5. すでにRDS MySQL インスタンスを作成済みです。

ステップ 1：MaxCompute に DLF および OSS へのアクセス権限を付与する

MaxCompute プロジェクトアカウントは、権限付与されていない場合、DLF および OSS にアクセスできません。以下のいずれかの方法で必要な権限を付与してください。

• ワンクリックによる権限付与：MaxCompute プロジェクトを作成したアカウントと DLF をデプロイしたアカウントが同一の場合、DLF の権限付与 をクリックして権限を付与します。

• カスタムによる権限付与：MaxCompute プロジェクトの作成と DLF のデプロイに同じアカウントを使用しているかどうかに関係なく、カスタム方法で DLF の権限を付与します。

ステップ 2：MySQL テストデータを準備する

すでに他の MySQL テストデータがある場合は、このステップをスキップできます。

1. RDS コンソールにログインします。

2. 左側のナビゲーションバーで インスタンス を選択し、左上隅でリージョンを選択します。

3. インスタンスページで、対象インスタンスの インスタンス ID/名前 をクリックして、インスタンス詳細ページを開きます。

4. 左側のナビゲーションバーで データベース管理 をクリックします。

5. データベースの作成 をクリックします。以下のパラメーターを設定します。

   パラメーター	必須	説明	例
   データベース (DB) 名	必須	名前は 2～64 文字である必要があります。 名前の先頭は英字、末尾は英字または数字である必要があります。 小文字の英字、数字、アンダースコア (_)、ハイフン (-) を使用できます。 インスタンス内でデータベース名は一意である必要があります。 データベース名に - が含まれる場合、作成されたデータベースのフォルダー名内の - は @002d に変更されます。	mysql_paimon
   サポートする文字セット	必須	必要に応じて文字セットを選択します。	utf8
   所属アカウント	任意	データベースの権限付与アカウントを選択します。このパラメーターは空のままにしておき、データベース作成後にアカウントをアタッチすることもできます。 ここには標準アカウントのみ表示されます。特権アカウントはすべてのデータベースに対してすべての権限を持っているため、権限付与は不要です。	Default
   注記	任意	データベース管理を容易にするために説明を入力します。説明は最大 256 文字まで入力できます。	Flink テスト用データベースを作成します。

6. データベースにログイン をクリックします。左側のナビゲーションバーで データベースインスタンス を選択し、作成したデータベースをダブルクリックします。右側の SQL コンソール ページで、次のステートメントを実行してテストテーブルを作成し、テストデータを書き込みます。

   インスタンスは存在するにもかかわらず、インスタンスを展開しても目的のデータベースが見つからない場合、考えられる原因は以下のとおりです。

   • ログインアカウントに目的のデータベースへのアクセス権限がない：RDS インスタンス詳細ページの アカウント管理 ページで、アカウント権限を手動で変更するか、ログインデータベースアカウントを変更してください。

   • メタデータが同期されていないためディレクトリが表示されない：インスタンス名にマウスカーソルを合わせ、インスタンス名の右側にある アイコンをクリックしてリストを更新し、データベースを表示します。

   -- テーブルを作成します。
   CREATE TABLE sales (
       id INT NOT NULL AUTO_INCREMENT,
       year INT NOT NULL,
       amount DECIMAL(10,2) NOT NULL,
       product_name VARCHAR(100) NOT NULL,
       customer_name VARCHAR(100) NOT NULL,
       order_date DATE NOT NULL,
       region VARCHAR(50) NOT NULL,
       status VARCHAR(20) NOT NULL,
       PRIMARY KEY (id,year)
   ) PARTITION BY RANGE (year) (
       PARTITION p2020 VALUES LESS THAN (2021),
       PARTITION p2021 VALUES LESS THAN (2022),
       PARTITION p2022 VALUES LESS THAN (2023),
       PARTITION p2023 VALUES LESS THAN (2024)
   );
   
   -- データを挿入します。
   INSERT INTO sales (year, amount, product_name, customer_name, order_date, region, status) VALUES
   (2020, 100.00, 'Product A', 'Customer 1', '2020-01-01', 'Region 1', 'Completed'),
   (2020, 200.00, 'Product B', 'Customer 2', '2020-02-01', 'Region 2', 'Pending'),
   (2021, 150.00, 'Product C', 'Customer 3', '2021-03-01', 'Region 3', 'Completed'),
   (2021, 300.00, 'Product D', 'Customer 4', '2021-04-01', 'Region 4', 'Pending'),
   (2022, 250.00, 'Product E', 'Customer 5', '2022-05-01', 'Region 5', 'Completed'),
   (2022, 400.00, 'Product F', 'Customer 6', '2022-06-01', 'Region 6', 'Pending'),
   (2023, 350.00, 'Product G', 'Customer 7', '2023-07-01', 'Region 7', 'Completed'),
   (2023, 500.00, 'Product H', 'Customer 8', '2023-08-01', 'Region 8', 'Pending'),
   (2020, 450.00, 'Product I', 'Customer 9', '2020-09-01', 'Region 1', 'Completed'),
   (2021, 600.00, 'Product J', 'Customer 10', '2021-10-01', 'Region 2', 'Pending');

7. テストテーブルのデータを照会します。

   SELECT * FROM sales;

   次のような結果が返されます。

   

ステップ 3：DLF メタデータベースを準備する

1. OSS コンソールにログインし、バケットを作成します。この例では、バケット名は mc-lakehouse-dlf-oss です。詳細については、「バケットの作成」をご参照ください。

2. バケット内に flink_paimon という名前のフォルダーを作成します。

3. Data Lake Formation (DLF) コンソールにログインし、左上隅でリージョンを選択します。

4. 左側のナビゲーションウィンドウで Metadata > Metadata を選択します。

5. Metadata ページで、Database タブをクリックします。

6. default Catalog List の下で、Create Database をクリックします。以下のパラメーターを設定します。

   パラメーター	必須	説明
   Catalog	必須	この例では、データカタログは default です。
   Database Name:	必須	カスタムのデータベース名を入力します。名前の先頭は英字で、長さは 1～128 文字で、英字、数字、アンダースコア (_) を使用できます。例：db_dlf_oss。
   Database Description:	任意	カスタムの説明を入力します。
   Select Path:	必須	データベースの保存場所です。例：oss://mc-lakehouse-dlf-oss/flink_paimon/。

ステップ 4：Flink を使用して Paimon カタログおよび MySQL カタログを作成する

1. Paimon カタログの作成：

   1. Apache Flink 用リアルタイムコンピュートコンソールにログインし、左上隅でリージョンを選択します。

   2. 対象ワークスペースの名前をクリックし、左側のナビゲーションウィンドウで Data Management を選択します。

   3. カタログリスト ページで、カタログの作成 をクリックします。カタログの作成 ダイアログボックスで、Apache Paimon を選択し、次へ をクリックして、以下のパラメーターを設定します。

      パラメーター	必須	説明
      metastore	必須	メタストアのタイプです。この例では、dlf を選択します。
      catalog name	必須	関連付ける DLF カタログのバージョンです。この例では、v1.0 を選択します。
      warehouse	必須	OSS 内のデータウェアハウスディレクトリです。この例では、ディレクトリは oss://mc-lakehouse-dlf-oss/flink_paimon/ です。
      fs.oss.endpoint	必須	OSS のエンドポイントです。たとえば、中国 (杭州) リージョンのエンドポイントは oss-cn-hangzhou-internal.aliyuncs.com です。
      fs.oss.accessKeyId	必須	OSS にアクセスするための AccessKey ID です。
      fs.oss.accessKeySecret	必須	OSS にアクセスするための AccessKey Secret です。
      dlf.catalog.accessKeyId	必須	DLF にアクセスするための AccessKey ID です。
      dlf.catalog.accessKeySecret	必須	DLF にアクセスするための AccessKey Secret です。

2. MySQL カタログの作成：

   1. Apache Flink 用リアルタイムコンピュートコンソールにログインし、左上隅でリージョンを選択します。

   2. ホワイトリストを追加できます。

      1. 対象ワークスペースの 操作 列で、詳細 をクリックします。

         表示される ワークスペースの詳細 ダイアログボックスで、VSwitch の CIDR ブロック をコピーします。

      2. RDS コンソールにログインします。

         左側のナビゲーションバーで インスタンス を選択し、左上隅でリージョンを選択します。

         インスタンスページで、対象インスタンスの インスタンス ID/名前 をクリックして、インスタンス詳細ページを開きます。

      3. 左側のナビゲーションバーで ホワイトリストとセキュリティグループ をクリックします。

         ホワイトリストの設定 タブで、編集 をクリックします。

      4. 表示される グループを変更する ダイアログボックスで、コピーした IP アドレスを IP アドレス フィールドに追加し、OK をクリックします。

   3. Apache Flink 用リアルタイムコンピュートコンソールにログインし、左上隅でリージョンを選択します。

      対象ワークスペースの名前をクリックし、左側のナビゲーションウィンドウで Data Management を選択します。

   4. カタログリスト ページで、カタログの作成 をクリックします。カタログの作成 ダイアログボックスで、MySQL を選択し、次へ をクリックして、以下のパラメーターを設定します。

      パラメーター	必須	説明
      catalog name	必須	MySQL カタログのカスタム名を入力します。例：mysql-catalog。
      hostname	必須	MySQL データベースの IP アドレスまたはホスト名です。 RDS MySQL コンソールにログインし、データベースインスタンスの詳細ページで データベースの接続 をクリックすると、内部エンドポイント、パブリックエンドポイント、および 内部ポート を確認できます。 クロス VPC または パブリックネットワーク 経由でのアクセスにはネットワーク接続性が必要です。詳細については、「ネットワーク接続性」をご参照ください。
      port	デフォルト	サーバーへの接続ポートです。デフォルトポートは 3306 です。
      default database	必須	デフォルトデータベースの名前です。例：mysql_paimon。
      username	必須	MySQL データベースサーバーへの接続ユーザー名です。RDS for MySQL コンソールにログインし、データベースインスタンスページで アカウント管理 をクリックしてユーザー名を確認します。
      password	必須	MySQL データベースサーバーへの接続パスワードです。RDS for MySQL コンソールにログインし、データベースインスタンスページで アカウント管理 をクリックしてパスワードを確認します。

ステップ 5：Flink を使用して MySQL からデータを読み取り、Paimon に書き込み、メタデータを DLF に同期する

1. Apache Flink 用リアルタイムコンピュートコンソールにログインし、左上隅でリージョンを選択します。

2. 対象のワークスペース名をクリックし、左側のナビゲーションウィンドウで [開発] > [ETL] を選択します。

3. 下書き タブで、 をクリックして新しいフォルダーを作成します。

4. フォルダーを右クリックし、空のストリーム下書きの新規作成 を選択します。下書きの新規作成 ダイアログボックスで、名前 を入力し、エンジンバージョン を選択します。

5. ファイルに次の CREATE TABLE AS (CTAS) ステートメントを記述します。コード内の名前を実際の名前に置き換えてください。

   CREATE TABLE IF NOT EXISTS `<dlf_meta_db_name>`.`<OSS_bucket_name>`.`sales` 
   AS TABLE `<mysql_catalog_name>`.`<RDS_mysql_name>`.`sales`;
   
   -- このトピックで使用している名前を使用する場合は、次のコードをコピーできます。
   CREATE TABLE IF NOT EXISTS `db_dlf_oss`.`flink_paimon`.`sales` 
   AS TABLE `mysql-catalog`.`mysql_paimon`.`sales`;

   1. （任意）右上隅の 検証 をクリックして、ジョブの Flink SQL ステートメントに構文エラーがないかをチェックします。

   2. 右上隅の デプロイ をクリックし、表示される 新バージョンのデプロイ ダイアログボックスで、コメント、ジョブラベル、および デプロイ先 を入力し、確認 をクリックします。

6. 対象のワークスペースの名前をクリックします。左側のナビゲーションウィンドウで、[O&M] > [デプロイメント]を選択します。

7. デプロイメント ページで、対象ジョブをクリックして デプロイメントの詳細 ページを開きます。

8. ページの右上隅で 開始 をクリックし、ステートレス起動 を選択して、開始。

9. Paimon データを照会します。

   左側のナビゲーションウィンドウで 開発 > データクエリ を選択します。

   クエリスクリプト タブで、 をクリックして新しいクエリスクリプトを作成します。

   次のコードを実行します。

   SELECT * FROM `<paimon_catalog_name>`.`flink_paimon`.`sales`;

   次のような結果が返されます。

   

10. OSS コンソールにアクセスし、mc-lakehouse-dlf-oss/flink_paimon/ ディレクトリを確認します。sales/ フォルダーが生成されています（下図参照）。

    

11. Data Lake Formation (DLF) コンソールにログインし、左上隅でリージョンを選択します。

    左側のナビゲーションウィンドウで Metadata > Metadata を選択します。

    flink_paimon というデータベース名をクリックします。生成されたテーブルが下図のように表示されます。

    

手順 6: MaxCompute で DLF+OSS 外部データソースを作成する

1. MaxCompute コンソールにログインし、左上隅でリージョンを選択します。

2. 左側のナビゲーションウィンドウで 設定管理 > 外部データソース を選びます。

3. 外部データソース ページで、外部データソースの作成 をクリックします。

4. 外部データソースの作成 ダイアログボックスで、パラメーターを設定します。以下の表にパラメーターを示します。

   パラメーター	必須	説明
   外部データソースタイプ	必須	DLF+OSS を選択します。
   外部データソース名	必須	カスタム名を入力します。命名規則は以下のとおりです。 名前の先頭は英字で、小文字の英字、アンダースコア (_)、数字のみを使用できます。 名前の長さは 128 文字以内である必要があります。 例：mysql_paimon_dlf。
   外部データソースの説明	任意	必要に応じて説明を入力します。
   リージョン	（必須）	現在のリージョンがデフォルトで選択されます。
   DLF エンドポイント	必須	現在のリージョンの DLF エンドポイントがデフォルトで使用されます。
   OSS エンドポイント	必須	現在のリージョンの OSS エンドポイントがデフォルトで使用されます。
   RoleARN	必須	RAM ロールの Alibaba Cloud リソースネーム (ARN) です。このロールには、DLF および OSS の両方にアクセスする権限が必要です。 RAM コンソールにログインします。 左側のナビゲーションバーで、アイデンティティ > ロール を選択します。 基本情報 セクションで、ARN を取得できます。 例：acs:ram::124****:role/aliyunodpsdefaultrole。
   外部データソースの補足プロパティ	任意	外部データソースの追加プロパティです。これらのプロパティを指定すると、この外部データソースを使用するタスクは、定義された動作に基づいてソースシステムにアクセスできます。 説明 サポートされているパラメーターについては、公式ドキュメントの今後のアップデートをご確認ください。

5. 確認 をクリックして、外部データソースを作成します。

6. 外部データソース ページで、データソースを見つけ、操作 列の 詳細 をクリックして詳細を表示します。

ステップ 7：外部スキーマを作成する

MaxCompute に接続し、次のコマンドを実行します。

SET odps.namespace.schema=true;

CREATE EXTERNAL SCHEMA IF NOT EXISTS <external_schema>
WITH <external_data_source>
ON '<dlf_data_catalogue>.dlf_database';

以下の表にパラメーターを示します。

• external_schema：外部スキーマの名前です。例：es_mc_dlf_oss_paimon。

• external_data_source：作成した外部データソースの名前です。外部スキーマを含むプロジェクトは、外部データソースと同じリージョンに配置されている必要があります。例：mysql_paimon_dlf。

• dlf_data_catalogue：DLF データカタログの ID です。データカタログの作成方法については、「データカタログの作成」をご参照ください。例：db_dlf_oss。

• dlf_database：指定された DLF データカタログ内のデータベース名です。詳細については、「データベース、テーブル、および関数」をご参照ください。例：flink_paimon。

ステップ 8：SQL を使用して OSS データにアクセスする

1. MaxCompute クライアントにログインし、外部スキーマ内のテーブルを照会します。

SET odps.namespace.schema=true;
use schema es_mc_dlf_oss_paimon;
SHOW tables IN es_mc_dlf_oss_paimon;

-- 次のような結果が返されます：
ALIYUN$xxx:sales

OK

2. 外部スキーマテーブル内のデータを照会します。

SET odps.namespace.schema=true;
SELECT * FROM <maxcompute_project_name>.es_mc_dlf_oss_paimon.sales;

-- 次のような結果が返されます：
+------------+------------+------------+--------------+---------------+------------+------------+------------+
| id         | year       | amount     | product_name | customer_name | order_date | region     | status     | 
+------------+------------+------------+--------------+---------------+------------+------------+------------+
| 1          | 2020       | 100        | Product A    | Customer 1    | 2020-01-01 | Region 1   | Completed  | 
| 2          | 2020       | 200        | Product B    | Customer 2    | 2020-02-01 | Region 2   | Pending    | 
| 3          | 2021       | 150        | Product C    | Customer 3    | 2021-03-01 | Region 3   | Completed  | 
| 4          | 2021       | 300        | Product D    | Customer 4    | 2021-04-01 | Region 4   | Pending    | 
| 5          | 2022       | 250        | Product E    | Customer 5    | 2022-05-01 | Region 5   | Completed  | 
| 6          | 2022       | 400        | Product F    | Customer 6    | 2022-06-01 | Region 6   | Pending    | 
| 7          | 2023       | 350        | Product G    | Customer 7    | 2023-07-01 | Region 7   | Completed  | 
| 8          | 2023       | 500        | Product H    | Customer 8    | 2023-08-01 | Region 8   | Pending    | 
| 9          | 2020       | 450        | Product I    | Customer 9    | 2020-09-01 | Region 1   | Completed  | 
| 10         | 2021       | 600        | Product J    | Customer 10   | 2021-10-01 | Region 2   | Pending    | 
+------------+------------+------------+--------------+---------------+------------+------------+------------+