:DLF、ApsaraDB RDS または Realtime Compute for Apache Flink、および OSS に基づく Delta Lake または Apache Hudi ストレージメカニズム

手順

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

   MaxCompute プロジェクトを管理する Alibaba Cloud アカウントに、DLF および OSS へのアクセス権限を付与します。

2. ステップ 2: OSS でバケットとフォルダを作成する

   データレイクの統一ストレージパスとして OSS バケットを作成します。

3. ステップ 3: データレイクに取り込むデータを準備する

   ApsaraDB RDS データベースを作成し、データレイクに取り込むデータを準備します。

4. ステップ 4: DLF にデータソースを追加し、メタデータベースを作成する

   ApsaraDB RDS を DLF のデータソースとして追加します。

5. ステップ 5: DLF コンソールでデータ取り込みタスクを作成して開始する

   DLF コンソールでデータ取り込みタスクを作成して、ApsaraDB RDS データベースからテーブルデータを同期し、データをリアルタイムでデータレイクに再生します。

6. ステップ 6: MaxCompute に基づいてデータレイクのデータを分析する

   DataWorks コンソールの Lake and Warehouse Integration (Data Lakehouse) ページで外部プロジェクトを作成して、データレイクのデータを分析します。

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

ステップ 2: OSS でバケットとフォルダを作成する

データレイクの統一ストレージパスとして OSS バケットを作成します。

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

2. 左側のナビゲーションウィンドウで、[バケット] をクリックします。 [バケット] ページで、[バケットの作成] をクリックします。

3. [バケットの作成] パネルで、[バケット名] パラメータを構成し、リージョンを選択して、[作成] をクリックします。 たとえば、バケット名パラメータを mc-dlf-oss に設定し、リージョン ドロップダウンリストから 中国 (上海) リージョンを選択できます。

   

4. [バケット] ページで、作成したバケットの名前をクリックして、[オブジェクト] ページに移動します。

5. ページの右側にある [ディレクトリの作成] をクリックします。 [ディレクトリの作成] パネルで、[ディレクトリ名] (例: datalaketest) を指定し、[OK] をクリックします。

   

ステップ 3: データレイクに取り込むデータを準備する

ApsaraDB RDS データベースを構築し、テーブルを作成し、データレイクに取り込むデータを準備します。

1. ApsaraDB RDS コンソールの [インスタンス] ページに移動します。 上部のナビゲーションバーで、中国 (上海) などのリージョンを選択します。 次に、インスタンスリストで使用したいインスタンスの ID をクリックします。 この例では、ApsaraDB RDS for MySQL インスタンスが使用されています。

2. ApsaraDB RDS データベースアカウントを作成します。 インスタンス詳細ページの左側のナビゲーションウィンドウで、[アカウント管理] をクリックします。 [ユーザーアカウント] タブで、[アカウントを作成する] をクリックします。 [アカウントの作成] パネルで、パラメータを構成し、[OK] をクリックします。 次の表にパラメータを示します。

   

   パラメータ	説明	例
   [データベースアカウント]	ApsaraDB RDS データベースへのアクセスに使用するアカウント。 この情報は、データベースの作成時に ApsaraDB RDS データベースに関連付けられます。	datalake_mcdbuser
   [アカウントタイプ]	アカウントのタイプ。 有効な値: [一般アカウント]: このアカウントを ApsaraDB RDS データベースに関連付ける必要があります。 [特権アカウント]: このアカウントは、ApsaraDB RDS インスタンスのすべてのデータベースに対する権限を持っています。 特権アカウントに特定のデータベースに対する権限を付与する必要はありません。	一般アカウント
   [新しいパスワード]	アカウントのパスワード。	なし
   [パスワードの確認]	アカウントのパスワードを確認します。	なし

   パラメータの詳細については、「ApsaraDB RDS for MySQL インスタンスのデータベースとアカウントを作成する」をご参照ください。

3. ApsaraDB RDS データベースを作成します。 インスタンス詳細ページの左側のナビゲーションウィンドウで、[データベース] をクリックします。 ページの右側にある [データベースの作成] をクリックします。「データベースの作成」パネルで、パラメータを構成し、「作成」をクリックします。 次の表にパラメータを示します。

   パラメータ	説明
   [データベース名]	名前は 2 ～ 64 文字の長さでなければなりません。 文字で始まり、文字または数字で終わる必要があります。 小文字、数字、アンダースコア (_)、またはハイフン (-) で構成できます。 データベース名はインスタンス内で一意である必要があります。 説明 データベース名にハイフン (-) が含まれている場合、作成されたデータベースの対応するフォルダ名は、ハイフン (-) を @002d に置き換えます。
   [サポートされている文字セット]	必要に応じて文字セットを選択します。
   [認可者]	このデータベースにアクセスする必要があるアカウントを選択します。 このパラメータは空のままにすることができ、データベースの作成後にアカウントをバインドできます。 詳細については、「アカウント権限を変更する」をご参照ください。 説明 高特権アカウントはすべてのデータベースに対するすべての権限を自動的に持つため、認可を必要としないため、ここに表示されるのは標準アカウントのみです。
   [説明]	オプション。 このパラメータは、後続のデータベース管理を容易にするために、データベースに関する注釈を追加するために使用されます。 最大 256 文字まで入力できます。

4. ApsaraDB RDS データベースにテーブルを作成し、少量のテストデータをテーブルに挿入します。 詳細については、「DMS を使用して ApsaraDB RDS for MySQL インスタンスにログインする」をご参照ください。 この例では、anti_fraud_result という名前のテーブルが作成されます。 サンプルステートメント:

   CREATE TABLE `anti_fraud_result` (
     `transactionid` varchar(32) NOT NULL,
     `uid` varchar(32) DEFAULT NULL,
     `card` varchar(32) DEFAULT NULL,
     `longitude` double(12,8) DEFAULT '12.00000000',
     `latitude` double(12,8) DEFAULT '12.00000000',
     PRIMARY KEY (`transactionid`)
   ) ENGINE=InnoDB DEFAULT CHARSET=utf8
   ;
   
   INSERT INTO `anti_fraud_result` values ("12489571","82005","123123",3.14592040,101.12315432);
   INSERT INTO `anti_fraud_result` values ("12489572","82005","123123",3.14592040,101.12315432);
   INSERT INTO `anti_fraud_result` values ("12489573","82005","123123",3.14592040,101.12315432);
   INSERT INTO `anti_fraud_result` values ("12489574","82005","123123",3.14592040,101.12315432);

(オプション) ステップ 4: DLF にデータソースを追加し、メタデータベースを作成する

DLF でデータレイクのデータを管理するためのメタデータベースを作成します。

1. DLF コンソール にログインします。 上部のナビゲーションバーで、中国 (上海) などのリージョンを選択します。

2. datalake メタデータベースなどのメタデータベースを作成します。

   1. DLF コンソールの左側のナビゲーションウィンドウで、[Metadata] > [Metadata]を選択します。 Metadataページの Database タブで、Create Database をクリックします。

   2. Create Database ダイアログボックスで、パラメータを構成します。 次の表にパラメータを示します。

      パラメータ	説明	例
      カタログ	データが保存されているデータベースカタログの名前。	default
      データベース名	ターゲット メタデータベースの名前。	dlf_db
      データベースの説明	作成するメタデータベースの説明。	データレイクハウス
      パスの選択	データファイルが保存されている OSS ディレクトリ。 ディレクトリは oss://<バケット名>/<OSS ディレクトリ名> 形式である必要があります。	oss://doc-test-01/datalake/

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

ステップ 5: DLF コンソールでデータ取り込みタスクを作成して開始する

DLF のデータ取り込み機能は更新されなくなりました。 次のいずれかの方法を使用して、データを Delta Lake ファイルまたは Apache Hudi ファイルとしてデータレイクに取り込むことができます。

ステップ 6：MaxCompute に基づいてデータレイクのデータを分析する

1. DataWorks コンソールで外部プロジェクトを作成します。

   1. DataWorks コンソール にログオンします。 上部のナビゲーションバーで、[中国 (上海)] リージョンを選択します。

   2. DataWorks コンソールの左側のナビゲーションウィンドウで、[その他] > [Lake And Warehouse Integration (Data Lakehouse)] を選択します。

   3. [Lake And Warehouse Integration (Data Lakehouse)] ページで、[開始] をクリックします。

   4. [Data Lakehouse の作成] ページで、パラメーターを構成します。 次の表にパラメーターを示します。

      表 1. データウェアハウスの作成手順のパラメーター

      パラメーター	説明
      外部プロジェクト名	ext_dlf_delta
      MaxCompute プロジェクト	ms_proj1

      表 2. データレイク接続の作成手順のパラメーター

      パラメーター		説明
      異種データプラットフォームタイプ	ドロップダウンリストから [Alibaba Cloud DLF + OSS] を選択します。	なし
      Alibaba Cloud DLF + OSS	外部プロジェクトの説明	なし
      	DLF がアクティブ化されているリージョン	cn-shanghai
      	DLF エンドポイント	dlf-share.cn-shanghai.aliyuncs.com
      	DLF データベース名	datalake
      	DLF RoleARN	なし

   5. [作成] をクリックします。 表示されるページで、[プレビュー] をクリックします。

      DLF データベースのテーブル情報をプレビューできる場合、操作は成功です。

   説明

   このステップでは、DataWorks コンソールで外部プロジェクトを作成する方法について説明します。 SQL 文を使用して外部プロジェクトを作成する方法の詳細については、「SQL 文を使用して外部プロジェクトを管理する」をご参照ください。

2. DataWorks コンソールの [アドホッククエリ] ページで、外部プロジェクトのテーブルを表示します。

   サンプル文：

   show tables in ext_dlf_delta;

   次の結果が返されます。

   ALIYUN$***@aliyun.com:anti_rds

   説明

   DataWorks の DataStudio によって提供されるアドホッククエリ機能の詳細については、「アドホッククエリノードを使用して SQL 文を実行する (オプション)」をご参照ください。

3. DataWorks コンソールの [アドホッククエリ] ページで、外部プロジェクトのテーブルデータをクエリします。

   説明

   クエリ結果が文字化けしている場合は、「バッチ同期」の「エンコード形式の構成の問題または文字化けによって発生するダーティデータエラーを処理するにはどうすればよいですか？」セクションの手順に従って問題を解決してください。

   • サンプル文:

     select * from ext_dlf_delta.anti_rds;

   • 次の図は、返された結果を示しています。

手順

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

   MaxCompute プロジェクトを管理する Alibaba Cloud アカウントに、DLF および OSS アクセス権限を付与します。

2. ステップ 2：OSS でバケットとフォルダを作成する

   データレイクの統合ストレージパスとして OSS バケットを作成します。

3. ステップ 3：データレイクに取り込むデータを準備する

   Realtime Compute for Apache Flink コンソールで一時テーブルを作成し、データレイクに取り込むデータを準備します。

4. ステップ 4：DLF にデータソースを追加し、メタデータベースを作成する

   DLF にデータソースを追加し、メタデータベースを作成します。

5. ステップ 5：DLF コンソールでメタデータ抽出タスクを作成して開始する

   DLF コンソールでメタデータ抽出タスクを作成し、OSS ディレクトリのテーブルデータをデータレイクに抽出します。

6. ステップ 6：MaxCompute に基づいてデータレイクのデータを分析する

   DataWorks コンソールの Lake and Warehouse Integration (Data Lakehouse) ページで外部プロジェクトを作成し、データレイクのデータを分析します。

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

ステップ 2：OSS でバケットとフォルダを作成する

データレイクの統合ストレージパスとして OSS バケットを作成します。

1. [OSS コンソール] にログインします。

2. 左側のナビゲーションウィンドウで、[バケット] をクリックします。[バケット] ページで、[バケットの作成] をクリックします。

3. [バケットの作成] パネルで、[バケット名] パラメータを設定し、リージョンを選択して、[作成] をクリックします。たとえば、「バケット名」パラメータを mc-dlf-oss に設定し、リージョンドロップダウンリストから 中国 (上海) リージョンを選択できます。

   

4. [バケット] ページで、作成したバケットの名前をクリックして、[オブジェクト] ページに移動します。

5. ページの右側で、[ディレクトリの作成] をクリックします。[ディレクトリの作成] パネルで、[ディレクトリ名] (例：datalaketest) を指定し、[OK] をクリックします。

   

ステップ 3：データレイクに取り込むデータを準備する

Hudi コネクタを使用して一時テーブルを作成し、Realtime Compute for Apache Flink コンソールでデータレイクに取り込むデータを準備します。詳細については、「SQL デプロイメントの概要」をご参照ください。

1. Realtime Compute for Apache Flink コンソールの [インスタンスリスト] に移動します。上部のナビゲーションバーで、Realtime Compute for Apache Flink インスタンスが配置されているリージョンを選択します。次に、インスタンスの ID をクリックします。

2. 左側のナビゲーションウィンドウで、[開発] > [ETL] を選択します。次に、[新規] をクリックして空のストリームドラフトを作成し、[次へ] をクリックします。

3. [新規ドラフト] ダイアログボックスで、ジョブ情報を指定し、[作成] をクリックします。

4. SQL エディターに次の文を入力します。構文の詳細については、「Hudi コネクタ (廃止予定)」をご参照ください。

   -- Create a temporary table named datagen as the data source. 
   CREATE TEMPORARY TABLE datagen(
     id INT NOT NULL PRIMARY KEY NOT ENFORCED,
     data  STRING,
     ts TIMESTAMP(3)
   ) WITH (
     'connector' = 'datagen' ,
     'rows-per-second'='100' 
   );
   
   -- Create a temporary table named flink_hudi_tbl as the result table. The data store points to OSS and the data is stored in the Apache Hudi format.
   CREATE TEMPORARY TABLE flink_hudi_tbl (
     id INT NOT NULL PRIMARY KEY NOT ENFORCED,
     data STRING,
     ts TIMESTAMP(3)
   ) WITH (
     'connector' = 'hudi', 
     'oss.endpoint' = 'oss-cn-beijing-internal.aliyuncs.com', 
     'accessKeyId' = '${secret_values.ak_id}', 
     'accessKeySecret' = '${secret_values.ak_secret}', 
     'path' = 'oss://<yourOSSBucket>/<Custom storage location>',
     'table.type' = 'MERGE_ON_READ',
     'hive_sync.enable' = 'true',
     'hive_sync.mode' = 'hms',
     'hive_sync.db' = 'flink_hudi',
     'hive_sync.table' = 'flink_hudi_tbl',
     'dlf.catalog.region' = 'cn-beijing',
     'dlf.catalog.endpoint' = 'dlf-vpc.cn-beijing.aliyuncs.com'
   );
   
   -- Write data from the source table to the result table.
   INSERT INTO flink_hudi_tbl SELECT * from datagen;

   パラメータ

   パラメータ	説明
   oss.endpoint	Flink インスタンスが存在するリージョンの内部エンドポイント。各リージョンのエンドポイントの詳細については、「リージョンとエンドポイント」をご参照ください。
   accessKeyId	Alibaba Cloud アカウントの AccessKey ID。
   accessKeySecret	Alibaba Cloud アカウントの AccessKey シークレット。
   path	OSS バケットのパス。
   dlf.catalog.region	Alibaba Cloud DLF のリージョン。詳細については、「サポートされているリージョンとエンドポイント」をご参照ください。
   dlf.catalog.endpoint	DLF のエンドポイント。詳細については、「サポートされているリージョンとエンドポイント」をご参照ください。 説明 このパラメーターは、DLF の VPC エンドポイントに設定することをお勧めします。中国 (杭州) リージョンを選択した場合は、このパラメーターを dlf-vpc.cn-hangzhou.aliyuncs.com に設定します。

5. 構文チェックを実行し、ドラフトをデプロイします。詳細については、「SQL デプロイメントの概要」のステップ 4 と 6 をご参照ください。

   1. SQL エディターページの右上隅にある [検証] をクリックして、構文チェックを実行します。構文チェックに合格すると、[成功] メッセージが表示されます。

   2. SQL エディターページの右上隅にある [デプロイ] をクリックします。「ドラフトのデプロイ」ダイアログボックスで、関連パラメータを設定し、[確認] をクリックします。

6. ドラフトのデプロイメントを開始し、起動結果を表示します。詳細については、「SQL デプロイメントの概要」のステップ 7 をご参照ください。

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

   2. 目的のデプロイメントを見つけ、[操作] 列の [開始] をクリックします。「ジョブの開始」ダイアログボックスで、[初期モード] を選択し、[開始] をクリックします。デプロイメントの [ステータス] が [実行中] に変わると、デプロイメントは想定どおりに実行されています。

   3. 「デプロイメント」ページで、計算結果を表示します。

7. デプロイメントが開始して一定期間実行された後、OSS コンソールにログインし、ディレクトリに書き込まれたデータファイルを表示します。

ステップ 4：DLF にデータソースを追加し、メタデータベースを作成する

DLF でデータレイクのデータを管理するためのメタデータベースを作成します。

1. [DLF コンソール] にログインします。上部のナビゲーションバーで、リージョンを選択します。

2. datalake メタデータベースなどのメタデータベースを作成します。

   1. DLF コンソールの左側のナビゲーションウィンドウで、[Metadata] > [Metadata]を選択します。 Metadataページの Database タブで、Create Database をクリックします。

   2. Create Database ダイアログボックスで、パラメータを設定します。次の表にパラメータを示します。

      パラメータ	説明	例
      カタログ	データが格納されているデータベースカタログの名前。	default
      データベース名	デスティネーションメタデータベースの名前。	dlf_db
      データベースの説明	作成するメタデータベースの説明。	Data lakehouse
      パスの選択	データファイルが保存されている OSS ディレクトリ。ディレクトリは oss://<バケット名>/<OSS ディレクトリ名> 形式である必要があります。	oss://doc-test-01/datalake/

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

ステップ 5：DLF コンソールでメタデータ抽出タスクを作成して開始する

1. [DLF コンソール] の左側のナビゲーションウィンドウで、[Metadata] > [Metadata Discovery] を選択します。

2. [Metadata Discovery] ページで、「Extraction Task」タブの [Create Extraction Task] をクリックします。

3. 「抽出タスクの作成」ページで、関連パラメータを設定します。詳細については、「メタデータディスカバリ」をご参照ください。

4. [保存して実行] をクリックします。

5. メタデータテーブルを表示します。抽出タスクの [実行の進捗状況] が 100% に達したら、左側のナビゲーションウィンドウの [メタデータ] をクリックします。[テーブルリスト] タブで、テーブルの詳細を表示します。

6. テーブルデータをクエリします。左側のナビゲーションウィンドウで、[データ探索] をクリックします。表示されるページで、SQL 文を実行してテーブルデータをクエリします。

ステップ 6：MaxCompute に基づいてデータレイクのデータを分析する

詳細については、このトピックの「ステップ 6：MaxCompute に基づいてデータレイクのデータを分析する」をご参照ください。