Realtime Compute for Apache Flink:マテリアライズドテーブル (ストリームとバッチを統合したレイクハウスの構築)

1. Paimon カタログの作成

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

   Paimon カタログの作成

   1. Realtime Compute 管理コンソールにログインします。

   2. 対象のワークスペースの [Actions] 列にある [Console] をクリックします。

   3. 左側メニューで [Data Management] を選択し、[Create Catalog] をクリックします。[Apache Paimon] を選択し、[Next] をクリックします。

      パラメーターの説明：

      設定項目	説明	備考
      metastore	メタストアのタイプ。	この例では、メタストアタイプとして filesystem を使用します。
      catalog name	Paimon カタログの名前。	カスタムの英語名を入力します。この例では paimon を使用します。
      warehouse	OSS 内のデータウェアハウスディレクトリ。	oss://<bucket>/<object> のフォーマットを使用します。各項目は次のとおりです。 <bucket>：OSS バケットの名前。 <object>：データが格納されているパス。 OSS 管理コンソールでバケットとオブジェクト名を確認します。
      fs.oss.endpoint	OSS の接続アドレス。	Flink と OSS が同じリージョンにある場合は、プライベートネットワークエンドポイントを使用します。それ以外の場合は、パブリックネットワークエンドポイントを使用します。詳細については、「リージョンとエンドポイント」をご参照ください。
      fs.oss.accessKeyId	OSS への読み書き権限を持つ Alibaba Cloud アカウントまたは RAM ユーザーの AccessKey ID。	取得方法については、「AccessKeyの作成」をご参照ください。プレーンテキストの認証情報が公開されるのを避けるため、代わりに変数を使用してください。詳細については、「変数の管理」をご参照ください。
      fs.oss.accessKeySecret	OSS への読み書き権限を持つ Alibaba Cloud アカウントまたは RAM ユーザーの AccessKey シークレット。

2. ユーザー行動ログテーブル ods_user_log と商品情報テーブル ods_dim_product を作成します。

   1. Realtime Compute 管理コンソールにログインします。

   2. 対象のワークスペースの [Actions] 列にある [Console] をクリックします。

   3. 左側メニューで、[Data Development] > [Data Query] を選択します。次のコードをコピーして貼り付け、ソーステーブルを作成します。

      この例では、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. 右上隅にある [Run] をクリックしてテーブルを作成します。

   5. 左側メニューで [Data Management] を選択し、Paimon カタログをクリックしてから [Refresh] をクリックして新しいテーブルを表示します。

3. 「シミュレーションデータ生成のためのFakerコネクタ」を使用してモックデータを生成し、Paimon テーブルに書き込みます。

   1. 左側メニューで、[Data Development] > [ETL] を選択します。

   2. [New] をクリックし、[Blank stream draft] を選択して [Next] をクリックし、次に [Create] をクリックします。

   3. 次のSQL文を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'   -- 毎秒 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'        -- 毎秒 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. 右上隅にある [Deploy] をクリックしてジョブをデプロイします。

   5. 左側メニューで、[Operation Center] > [Job O&M] を選択します。対象のジョブの [Actions] 列にある [Start] をクリックし、[Stateless start] を選択してから [Start] をクリックします。

4. シミュレーションデータをクエリします。

   左側メニューで、[Data Development] > [Data Query] を選択します。次のSQL文をSQLエディターにコピーし、右上隅にある [Run] をクリックします。

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

データプレビュー

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

1. 左側メニューで、[Operation Center] > [Data Lineage] を選択し、対象のマテリアライズドテーブルを検索します。

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

3. マテリアライズドテーブルの [Data Preview] タブで、[Query] アイコンをクリックします。

データクエリ

左側メニューで、[Data Development] > [Data Query] を選択します。次のSQL文をSQLエディターにコピーし、コードスニペットを選択して [Run] をクリックして、dws_overall マテリアライズドテーブルをクエリします。

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