Realtime Compute for Apache Flink:マテリアライズドテーブルのクイックスタート

1. （任意）Paimon カタログを作成します。

   マテリアライズドテーブル機能は Apache Paimon を基盤としています。メタストアタイプが Filesystem の Paimon カタログを作成する必要があります。すでに作成済みの場合は、このステップをスキップできます。詳細については、「Paimon カタログの作成」をご参照ください。

   Paimon カタログの作成

   1. Realtime Compute for Apache Flink コンソール にログインします。

   2. 対象のワークスペースの[アクション] 列で、[コンソール] をクリックします。

   3. 左側のナビゲーションウィンドウで、[Data Management] を選択し、[カタログの作成] をクリックします。次に、[Apache Paimon] を選択し、[次へ] をクリックします。

      

      パラメーターの説明：

      パラメーター	説明	備考
      metastore	メタストアのタイプです。	本例では、Filesystem を使用します。
      catalog name	Paimon カタログの名前です。	任意の名前を入力します。本例では `paimon` を使用します。
      warehouse	Object Storage Service (OSS) 上のデータウェアハウスディレクトリです。	形式は oss://<bucket>/<object> です。ここで： <bucket>：ご利用の OSS バケット名です。 `` は、データが格納されているパスです。 OSS コンソールでバケット名およびオブジェクト名を確認できます。OSS コンソール
      fs.oss.endpoint	OSS サービスのエンドポイントです。	Flink と OSS が同一リージョンにある場合は、内部エンドポイントを使用します。それ以外の場合は、パブリックエンドポイントを使用します。詳細については、「リージョンとエンドポイント」をご参照ください。
      fs.oss.accessKeyId	OSS に対する読み取り／書き込み権限を持つ Alibaba Cloud アカウントまたは RAM ユーザーの AccessKey ID です。	AccessKey ペアの取得方法については、「AccessKey ペアの作成」をご参照ください。AccessKey ペアがプレーンテキストで公開されるのを防ぐため、変数の使用を推奨します。変数の管理方法については、「変数の管理」をご参照ください。
      fs.oss.accessKeySecret	OSS に対する読み取り／書き込み権限を持つ Alibaba Cloud アカウントまたは RAM ユーザーの AccessKey Secret です。

2. `ods_user_log` および `ods_dim_product` テーブルを作成します。

   1. Realtime Compute for Apache Flink コンソール にログインします。

   2. 対象のワークスペースで、[コンソール] を [操作] 列でクリックします。

   3. 左側ナビゲーションウィンドウで、Data Studio > Scripts を選択します。以下のコードをコピー＆ペーストして、ソーステーブルを作成します。

      本例では、`paimon` という名前の Paimon カタログおよびデフォルトデータベースを使用します。

      CREATE TABLE `paimon`.`default`.`ods_user_log` (
        item_id INT NOT NULL,
        user_id INT NOT NULL,
        vtime TIMESTAMP(6),
        ds VARCHAR(10) NOT NULL
      ) 
      PARTITIONED BY(ds)
      WITH (
        'bucket' = '4',            -- 4 つのバケットを指定。
        'bucket-key' = 'item_id'   -- バケットキーを指定。同じ item_id を持つデータは同一バケットに配置されます。
      );
      
      CREATE TABLE `paimon`.`default`.`ods_dim_product` (
        item_id INT NOT NULL,
        title VARCHAR(255),
        pict_url VARCHAR(255), 
        brand_id INT,
        seller_id INT,
        PRIMARY KEY(item_id) NOT ENFORCED
      ) WITH (
        'bucket' = '4',
        'bucket-key' = 'item_id'
      );

   4. 右上隅の[実行]をクリックして、対応するデータテーブルを作成します。

   5. 左側のナビゲーションウィンドウで、[Data Management] を選択します。次に、対応する Paimon カタログをクリックし、[Refresh] をクリックして、新しいテーブルを表示します。

3. Faker データジェネレーター コネクタを使用して、シミュレートされたデータを生成し、Paimon テーブルへ書き込みます。

   1. 左側ナビゲーションウィンドウで、Data Studio > ETL を選択します。

   2. ［新規］をクリックし、［空のストリーム下書き］を選択して、［次へ］をクリックし、次に［作成］をクリックします。

   3. 以下の SQL 文をエディターにコピーします。

      CREATE TEMPORARY TABLE `user_log` (
        item_id INT,  // 製品 ID
        user_id INT,  // ユーザー ID
        vtime TIMESTAMP,  
        ds AS DATE_FORMAT(CURRENT_DATE,'yyyyMMdd')
      ) WITH (
        'connector' = 'faker',    -- Faker コネクタ
        'fields.item_id.expression'='#{number.numberBetween ''0'',''1000''}',    -- 0 ～ 1000 の乱数を生成。
        'fields.user_id.expression'='#{number.numberBetween ''0'',''100''}',
        'fields.vtime.expression'='#{date.past ''5'',''HOURS''}',           -- 現在時刻から過去 5 時間分のデータを生成。
        'rows-per-second' = '3'   -- 1 秒あたり 3 行を生成。
       );
        
       CREATE TEMPORARY TABLE `dim_product` (
        item_id INT NOT NULL,
        title VARCHAR(255),
        pict_url VARCHAR(255), 
        brand_id INT,
        seller_id INT,
        PRIMARY KEY(item_id) NOT ENFORCED
       ) WITH (
        'connector' = 'faker',    -- Faker コネクタ
        'fields.item_id.expression'='#{number.numberBetween ''0'',''1000''}',
        'fields.title.expression'='#{book.title}',
        'fields.pict_url.expression'='#{internet.domainName}',
        'fields.brand_id.expression'='#{number.numberBetween ''1000'',''10000''}',   
        'fields.seller_id.expression'='#{number.numberBetween ''1000'',''10000''}',
        'rows-per-second' = '3'        -- 1 秒あたり 3 行を生成。
       );
      
      BEGIN STATEMENT SET; 
      
      INSERT INTO `paimon`.`default`.`ods_user_log` 
        SELECT 
        item_id,
        user_id,
        vtime,
        CAST(ds AS VARCHAR(10)) AS ds
      FROM `user_log`;
      INSERT INTO `paimon`.`default`.`ods_dim_product`
        SELECT 
        item_id,
        title,
        pict_url,
        brand_id,
        seller_id
      FROM `dim_product`;
      
      END; 

   4. 右上隅で、[デプロイ]をクリックします。

   5. 左側のナビゲーションウィンドウで、[オペレーションセンター] > [デプロイメント] をクリックします。対象のデプロイメントについて、[操作] 列の [開始] をクリックし、[ステートレス開始] を選択してから、[開始] をクリックします。

4. シミュレートされたデータをクエリします。

   左側のナビゲーションウィンドウで、[データ開発] > [データクエリ] を選択します。次のSQL文をSQL エディターにコピーし、右上隅の[実行]をクリックします。

   SELECT * FROM `paimon`.`default`.ods_dim_product LIMIT 10;
   
   SELECT * FROM `paimon`.`default`.ods_user_log LIMIT 10;

   

データのプレビュー

マテリアライズドテーブルの最新 100 行のデータをプレビューできます。

1. 左側のナビゲーションウィンドウで、[データ系統] をクリックして、対象のマテリアライズドテーブルを検索できます。

2. 対象のマテリアライズドテーブルのビューをクリックした後、ページの右下の隅にある[詳細]をクリックします。

3. 「[データプレビュー]」タブで、[クエリ] アイコンをクリックします。

   

データのクエリ

左側のナビゲーションウィンドウで [データ開発] > [データクエリ] に移動し、次の SQL 文を SQL エディターにコピーしてコードスニペットを選択してから、[実行] をクリックして dws_overall マテリアライズドテーブルにクエリを実行します。

SELECT * FROM `paimon`.`default`.dws_overall ORDER BY hh;