Paimon+StarRocks流式湖倉構建 - Realtime Compute for Apache Flink

背景資訊

隨著社會數字化發展，企業對資料時效性的需求越來越強烈。傳統的離線數倉搭建方法論比較明確，通過定時調度離線作業的方式，將上一時段產生的新鮮變更併入分層的數倉中（ODS->DWD->DWS->ADS），但是存在延時間長度和成本高兩大問題。離線作業的調度通常每小時甚至每天才進行一次，資料的消費者僅能看到上一小時甚至昨天的資料。同時，資料的更新多以覆寫（overwrite）分區的方式進行，需要重新讀取分區中原有的資料，才能與新鮮變更合并，產生新的結果資料。

基於Realtime ComputeFlink版和流式資料湖倉Paimon搭建流式湖倉可以解決上述傳統離線數倉的問題。利用Flink的Realtime Compute能力，資料可以在數倉分層之間即時資料流動。同時，利用Paimon高效的更新能力，資料變更可以在分鐘級的延時內傳遞給下遊消費者。因此，流式湖倉在延時和成本上具有雙重優勢。

關於流式資料湖倉Paimon的更多特性，請參見特色功能和Apache Paimon官方網站。

方案架構和優勢

架構

Realtime ComputeFlink版是強大的流式計算引擎，支援對海量即時資料高效處理。流式資料湖倉Paimon是流批統一的湖儲存格式，支援高吞吐的更新和低延後查詢。Paimon與Flink深度整合，能夠提供一體化的流式湖倉聯合解決方案。本文基於Flink+Paimon搭建流式湖倉的方案架構如下：

Flink將資料來源寫入Paimon，形成ODS層。
Flink訂閱ODS層的變更資料（Changelog）進行加工，形成DWD層再次寫入Paimon。
Flink訂閱DWD層的Changelog進行加工，形成DWS層再次寫入Paimon。
最後由開源巨量資料平台E-MapReduce的StarRocks讀取Paimon外部表格，對外提供應用查詢。

優勢

該方案有如下優勢：

Paimon的每一層資料都可以在分鐘級的延時內將變更傳遞給下遊，將傳統離線數倉的延時從小時級甚至天級降低至分鐘級。
Paimon的每一層資料都可以直接接受變更資料，無需覆寫分區，極大地降低了傳統離線數倉資料更新與訂正的成本，解決了中介層資料不易查、不易更新、不易修正的問題。
模型統一，架構簡化。ETL鏈路的邏輯是基於Flink SQL實現的；ODS層、DWD層和DWS層的資料統一儲存在Paimon中，可以降低架構複雜度，提高資料處理效率。

該方案依賴於Paimon的三個核心能力，詳情如下表所示。

Paimon核心能力	詳情
主鍵表更新	Paimon底層使用LSM Tree資料結構，可以實現高效的資料更新。關於Paimon主鍵表、Paimon底層資料結構的介紹請參見Primary Key Table和File Layouts。
增量資料產生機制（Changelog Producer）	Paimon可以為任意輸入資料流產生完整的增量資料（所有的update_after資料都有對應的update_before資料），保證資料變更可以完整地傳遞給下遊。詳情請參見增量資料產生機制。
資料合併機制（Merge Engine）	當Paimon主鍵表收到多條具有相同主鍵的資料時，為了保持主鍵的唯一性，Paimon結果表會將這些資料合併成一條資料。Paimon支援去重、部分更新、預彙總等豐富多樣的資料合併行為，詳情請參見資料合併機制。

實踐情境

本文以某個電商平台為例，通過搭建一套流式湖倉，實現資料的加工清洗，並支援上層應用對資料的查詢。流式湖倉實現了資料的分層和複用，並支撐各個業務方的報表查詢（交易大屏、行為資料分析、使用者畫像標籤）以及個人化推薦等多個業務情境。

構建ODS層：業務資料庫即時入倉
MySQL有orders（訂單表），orders_pay（訂單支付表）和product_catalog（商品類別字典表）三張業務表，這三張表通過Flink即時寫入OSS，並以Paimon格式進行儲存，作為ODS層。
構建DWD層：主題寬表
將訂單表、商品類別字典表、訂單支付表利用Paimon的部分更新（partial-update）合并機制進行打寬，以分鐘級延時產生DWD層寬表併產出變更資料（Changelog）。
構建DWS層：指標計算
Flink即時消費寬表的變更資料，利用Paimon的預彙總（aggregation）合并機制產出DWM層dwm_users_shops（使用者-商戶彙總中間表），並最終產出DWS層dws_users（使用者彙總指標表）以及dws_shops（商戶彙總指標表）。

前提條件

已開通資料湖構建，推薦使用DLF 2.5提供作為儲存服務，詳情請參見快速使用DLF。
已開通Flink全託管，詳情請參見開通Realtime ComputeFlink版。
已開通EMR的StarRocks，詳情請參見快速使用存算分離版執行個體。

說明

StarRocks執行個體、DLF需要與Flink工作空間處於相同地區。

使用限制

僅Realtime Compute引擎VVR 11.1.0及以上版本支援該流式湖倉方案。

構建流式湖倉

準備MySQL CDC資料來源

本文以雲資料庫RDS MySQL版為例，建立資料庫名稱為order_dw，並建立三張業務表及資料。

（廢棄，重新導向到“第一步”）快速建立RDS MySQL執行個體。

重要
RDS MySQL版執行個體需要與Flink工作空間處於同一VPC。不在同一VPC下時請參見如何訪問跨VPC的其他服務？

（廢棄，重新導向至“第一步”）建立資料庫和帳號。

建立名稱為order_dw的資料庫，並建立高許可權帳號或具有資料庫order_dw讀寫權限的普通帳號。

建立三張表，並插入相應資料。

CREATE TABLE `orders` (
  order_id bigint not null primary key,
  user_id varchar(50) not null,
  shop_id bigint not null,
  product_id bigint not null,
  buy_fee bigint not null,   
  create_time timestamp not null,
  update_time timestamp not null default now(),
  state int not null
);
CREATE TABLE `orders_pay` (
  pay_id bigint not null primary key,
  order_id bigint not null,
  pay_platform int not null, 
  create_time timestamp not null
);
CREATE TABLE `product_catalog` (
  product_id bigint not null primary key,
  catalog_name varchar(50) not null
);
-- 準備資料
INSERT INTO product_catalog VALUES(1, 'phone_aaa'),(2, 'phone_bbb'),(3, 'phone_ccc'),(4, 'phone_ddd'),(5, 'phone_eee');
INSERT INTO orders VALUES
(100001, 'user_001', 12345, 1, 5000, '2023-02-15 16:40:56', '2023-02-15 18:42:56', 1),
(100002, 'user_002', 12346, 2, 4000, '2023-02-15 15:40:56', '2023-02-15 18:42:56', 1),
(100003, 'user_003', 12347, 3, 3000, '2023-02-15 14:40:56', '2023-02-15 18:42:56', 1),
(100004, 'user_001', 12347, 4, 2000, '2023-02-15 13:40:56', '2023-02-15 18:42:56', 1),
(100005, 'user_002', 12348, 5, 1000, '2023-02-15 12:40:56', '2023-02-15 18:42:56', 1),
(100006, 'user_001', 12348, 1, 1000, '2023-02-15 11:40:56', '2023-02-15 18:42:56', 1),
(100007, 'user_003', 12347, 4, 2000, '2023-02-15 10:40:56', '2023-02-15 18:42:56', 1);
INSERT INTO orders_pay VALUES
(2001, 100001, 1, '2023-02-15 17:40:56'),
(2002, 100002, 1, '2023-02-15 17:40:56'),
(2003, 100003, 0, '2023-02-15 17:40:56'),
(2004, 100004, 0, '2023-02-15 17:40:56'),
(2005, 100005, 0, '2023-02-15 18:40:56'),
(2006, 100006, 0, '2023-02-15 18:40:56'),
(2007, 100007, 0, '2023-02-15 18:40:56');

管理中繼資料

建立Paimon Catalog

登入Realtime Compute控制台。
在左側導覽列，選擇中繼資料管理頁面，單擊建立Catalog。
在內建Catalog頁簽，單擊Apache Paimon，單擊下一步。

填寫以下參數，選擇DLF作為儲存類型，單擊確定。

配置項

說明

是否必填

備忘

metastore

中繼資料存放區類型。

是

此樣本選擇為dlf儲存類型。

catalog name

DLF資料目錄名稱。

重要

使用RAM使用者或角色時，請確保擁有DLF資料讀寫權限，詳情請參見資料授權管理。

是

推薦使用DLF 2.5，無需您再填寫AccessKey等資訊，支援快速選擇已建立的DLF資料目錄，建立資料目錄操作請參見資料目錄。

通過建立資料目錄操作建立paimoncatalog後，選擇名稱為paimoncatalog的資料目錄。

在資料目錄下建立相應的order_dw資料庫，以便後續同步MySQL中order_dw庫下所有表的資料。

在左側導覽列，選擇資料查詢 > 查詢指令碼，單擊建立一個臨時查詢。
```
-- 使用paimoncatalog資料來源
USE CATALOG paimoncatalog;
-- 建立order_dw資料庫
CREATE DATABASE order_dw;
```
返回The following statement has been executed successfully!表示建立庫成功。

關於Paimon Catalog的更多使用方法詳情請參見管理Paimon Catalog。

建立MySQL Catalog

在中繼資料管理頁面，單擊建立Catalog。
在內建Catalog頁簽，單擊MySQL，單擊下一步。

填寫以下參數，單擊確定，建立名為mysqlcatalog的MySQL Catalog。

配置項	說明	是否必填	備忘
catalog name	Catalog名稱。	是	填寫為自訂的英文名。本文以mysqlcatalog為例。
hostname	MySQL資料庫的IP地址或者Hostname。	是	詳情請參見查看和管理執行個體串連地址和連接埠。由於RDS MySQL版執行個體和Flink全託管處於相同VPC，此處應填寫內網地址。
port	MySQL資料庫服務的連接埠號碼，預設值為3306。	否	詳情請參見查看和管理執行個體串連地址和連接埠。
default-database	預設的MySQL資料庫名稱。	是	本文填寫需要同步的資料庫名order_dw。
username	MySQL資料庫服務的使用者名稱。	是	本文為準備MySQL CDC資料來源中建立的帳號。
password	MySQL資料庫服務的密碼。	是	本文為準備MySQL CDC資料來源中建立的密碼。

構建ODS層：業務資料庫即時入倉

基於Flink CDC，通過資料攝入YAML作業實現MySQL資料同步至Paimon，一次性將ODS層構建出來。

建立並啟動資料攝入YAML同步作業。

在Realtime Compute控制台的資料開發 > 資料攝入頁面，建立名為ods的YAML空白草稿作業。

將如下代碼複製到編輯器，注意修改相應的使用者名稱和密碼等參數。

source:
  type: mysql
  name: MySQL Source
  hostname: rm-bp1e********566g.mysql.rds.aliyuncs.com
  port: 3306
  username: ${secret_values.username}
  password: ${secret_values.password}
  tables: order_dw.\.*  # 支援正則表達，讀取order_dw庫下所有的表
  #（可選）同步增量階段新建立的表的資料
  scan.binlog.newly-added-table.enabled: true
  #（可選）同步表注釋和欄位注釋
  include-comments.enabled: true
  #（可選）優先分發無界的分區以避免可能出現的TaskManager OutOfMemory問題
  scan.incremental.snapshot.unbounded-chunk-first.enabled: true
  #（可選）開啟解析過濾，加速讀取
  scan.only.deserialize.captured.tables.changelog.enabled: true
sink:
  type: paimon
  name: Paimon Sink
  catalog.properties.metastore: rest
  catalog.properties.uri: http://cn-beijing-vpc.dlf.aliyuncs.com
  catalog.properties.warehouse: paimoncatalog
  catalog.properties.token.provider: dlf
pipeline:
  name: MySQL to Paimon Pipeline

配置項	描述	是否必填	樣本
`catalog.properties.metastore`	Metastore類型，固定為rest。	是	rest
`catalog.properties.token.provider`	Token提供方，固定為dlf。	是	dlf
`catalog.properties.uri`	訪問DLF Rest Catalog Server的URI，格式為`http://[region-id]-vpc.dlf.aliyuncs.com`。詳見地區與服務存取點中的Region ID。	是	http://cn-beijing-vpc.dlf.aliyuncs.com
`catalog.properties.warehouse`	DLF Catalog名稱。	是	paimoncatalog
`hostname`	MySQL資料庫的IP地址或者Hostname。詳情請參見查看和管理執行個體串連地址和連接埠。由於RDS MySQL版執行個體和Flink全託管處於相同VPC，此處應填寫內網地址。	是	rm-bp1e********566g.mysql.rds.aliyuncs.com
`username`	MySQL資料庫使用者名稱。建議使用密鑰管理，詳情請參見變數管理。	是	${secret_values.username}
`password`	MySQL資料庫密碼。建議使用密鑰管理，詳情請參見變數管理。	是	${secret_values.password}

Paimon寫入效能最佳化請參見Paimon效能最佳化。

單擊右上方的部署。
在營運中心 > 作業營運，單擊剛剛部署的ods作業操作列的啟動，選擇無狀態啟動啟動作業。作業啟動配置詳情請參見作業啟動。

查看MySQL同步到Paimon的三張表的資料。

在Realtime Compute控制台的資料開發 > 資料查詢頁面的查詢指令碼頁簽，將如下代碼拷貝到查詢指令碼後，選中目標片段後單擊右上方的運行。
```
SELECT * FROM paimoncatalog.order_dw.orders ORDER BY order_id;
```
執行查詢後，返回 orders 表中的 7 條訂單記錄，包含 order_id（100001~100007）、user_id、shop_id、product_id、buy_fee（1000~5000）、create_time、update_time 和 state 列，state 值均為 1。

構建DWD層：主題寬表

建立DWD層Paimon寬表dwd_orders

在Realtime Compute控制台的資料開發 > 資料查詢頁面的查詢指令碼頁簽，將如下代碼拷貝到查詢指令碼後，選中目標片段後單擊右上方的運行。

CREATE TABLE paimoncatalog.order_dw.dwd_orders (
    order_id BIGINT,
    order_user_id STRING,
    order_shop_id BIGINT,
    order_product_id BIGINT,
    order_product_catalog_name STRING,
    order_fee BIGINT,
    order_create_time TIMESTAMP,
    order_update_time TIMESTAMP,
    order_state INT,
    pay_id BIGINT,
    pay_platform INT COMMENT 'platform 0: phone, 1: pc',
    pay_create_time TIMESTAMP,
    PRIMARY KEY (order_id) NOT ENFORCED
) WITH (
    'merge-engine' = 'partial-update', -- 使用部分更新資料合併機制產生寬表
    'changelog-producer' = 'lookup' -- 使用lookup增量資料產生機制以低延時產出變更資料
);

返回The following statement has been executed successfully!表示建立成功。

即時消費ODS層orders、orders_pay表的變更資料

在Realtime Compute控制台的資料開發 > ETL頁面，建立名為dwd的SQL流作業，並將如下代碼複製到SQL編輯器後，部署作業並無狀態啟動作業。

通過該SQL作業，orders表會與product_catalog表進行維表關聯，關聯後的結果將與orders_pay一起寫入dwd_orders表中，利用Paimon表的部分更新資料合併機制，將orders表和orders_pay表中order_id相同的資料進行打寬。

SET 'execution.checkpointing.max-concurrent-checkpoints' = '3';
SET 'table.exec.sink.upsert-materialize' = 'NONE';
SET 'execution.checkpointing.interval' = '10s';
SET 'execution.checkpointing.min-pause' = '10s';
-- Paimon目前暫不支援在同一個作業裡通過多條INSERT語句寫入同一張表，因此這裡使用UNION ALL。
INSERT INTO paimoncatalog.order_dw.dwd_orders 
SELECT 
    o.order_id,
    o.user_id,
    o.shop_id,
    o.product_id,
    dim.catalog_name,
    o.buy_fee,
    o.create_time,
    o.update_time,
    o.state,
    NULL,
    NULL,
    NULL
FROM
    paimoncatalog.order_dw.orders o 
    LEFT JOIN paimoncatalog.order_dw.product_catalog FOR SYSTEM_TIME AS OF proctime() AS dim
    ON o.product_id = dim.product_id
UNION ALL
SELECT
    order_id,
    NULL,
    NULL,
    NULL,
    NULL,
    NULL,
    NULL,
    NULL,
    NULL,
    pay_id,
    pay_platform,
    create_time
FROM
    paimoncatalog.order_dw.orders_pay;

查看寬表dwd_orders的資料

在Realtime Compute控制台的資料開發 > 資料查詢頁面的查詢指令碼頁簽，將如下代碼拷貝到查詢指令碼後，選中目標片段後單擊右上方的運行。
```
SELECT * FROM paimoncatalog.order_dw.dwd_orders ORDER BY order_id;
```
執行成功後，返回 dwd_orders 寬表的 7 條訂單記錄，包含 order_id、order_user_id、order_shop_id、order_product_id、order_product_catalog_name、order_fee、order_create_time、order_update_time 共 8 個欄位，order_id 範圍為 100001～100007，訂單金額從 1000 到 5000，建立時間均為 2023-02-15。

構建DWS層：指標計算

建立DWS層的彙總表dws_users以及dws_shops

在Realtime Compute控制台的資料開發 > 資料查詢頁面的查詢指令碼頁簽，將如下代碼拷貝到查詢指令碼後，選中目標片段後單擊右上方的運行。

-- 使用者維度彙總指標表。
CREATE TABLE paimoncatalog.order_dw.dws_users (
    user_id STRING,
    ds STRING,
    paid_buy_fee_sum BIGINT COMMENT '當日完成支付的總金額',
    PRIMARY KEY (user_id, ds) NOT ENFORCED
) WITH (
    'merge-engine' = 'aggregation', -- 使用預彙總資料合併機制產生彙總表
    'fields.paid_buy_fee_sum.aggregate-function' = 'sum' -- 對 paid_buy_fee_sum 的資料求和產生彙總結果
    -- 由於dws_users表不再被下遊流式消費，因此無需指定增量資料產生機制
);
-- 商戶維度彙總指標表。
CREATE TABLE paimoncatalog.order_dw.dws_shops (
    shop_id BIGINT,
    ds STRING,
    paid_buy_fee_sum BIGINT COMMENT '當日完成支付總金額',
    uv BIGINT COMMENT '當日不同購買使用者總人數',
    pv BIGINT COMMENT '當日購買使用者總人次',
    PRIMARY KEY (shop_id, ds) NOT ENFORCED
) WITH (
    'merge-engine' = 'aggregation', -- 使用預彙總資料合併機制產生彙總表
    'fields.paid_buy_fee_sum.aggregate-function' = 'sum', -- 對 paid_buy_fee_sum 的資料求和產生彙總結果
    'fields.uv.aggregate-function' = 'sum', -- 對 uv 的資料求和產生彙總結果
    'fields.pv.aggregate-function' = 'sum' -- 對 pv 的資料求和產生彙總結果
    -- 由於dws_shops表不再被下遊流式消費，因此無需指定增量資料產生機制
);
-- 為了同時計算使用者視角的彙總表以及商戶視角的彙總表，另外建立一個以使用者 + 商戶為主鍵的中間表。
CREATE TABLE paimoncatalog.order_dw.dwm_users_shops (
    user_id STRING,
    shop_id BIGINT,
    ds STRING,
    paid_buy_fee_sum BIGINT COMMENT '當日使用者在商戶完成支付的總金額',
    pv BIGINT COMMENT '當日使用者在商戶購買的次數',
    PRIMARY KEY (user_id, shop_id, ds) NOT ENFORCED
) WITH (
    'merge-engine' = 'aggregation', -- 使用預彙總資料合併機制產生彙總表
    'fields.paid_buy_fee_sum.aggregate-function' = 'sum', -- 對 paid_buy_fee_sum 的資料求和產生彙總結果
    'fields.pv.aggregate-function' = 'sum', -- 對 pv 的資料求和產生彙總結果
    'changelog-producer' = 'lookup', -- 使用lookup增量資料產生機制以低延時產出變更資料
    -- dwm層的中間表一般不直接提供上層應用查詢，因此可以針對寫入效能進行最佳化。
    'file.format' = 'avro', -- 使用avro行存格式的寫入效能更加高效。
    'metadata.stats-mode' = 'none' -- 放棄統計資訊會增加OLAP查詢代價（對持續的流處理無影響），但會讓寫入效能更加高效。
);

返回The following statement has been executed successfully!表示建立成功。

DWD層dwd_orders表的變更資料

在Realtime Compute控制台資料開發 > ETL頁簽，建立名為dwm的SQL流作業，並將如下代碼複製到SQL編輯器後，部署作業並無狀態啟動作業。

通過該SQL作業，dwd_orders表的資料會寫入dwm_users_shops表中，利用Paimon表的預彙總資料合併機制，自動對order_fee求和，算出使用者在商戶的消費總額。同時，自動對1求和，也能算出使用者在商戶的消費次數。

SET 'execution.checkpointing.max-concurrent-checkpoints' = '3';
SET 'table.exec.sink.upsert-materialize' = 'NONE';
SET 'execution.checkpointing.interval' = '10s';
SET 'execution.checkpointing.min-pause' = '10s';
INSERT INTO paimoncatalog.order_dw.dwm_users_shops
SELECT
    order_user_id,
    order_shop_id,
    DATE_FORMAT (pay_create_time, 'yyyyMMdd') as ds,
    order_fee,
    1 -- 一條輸入記錄代表一次消費
FROM paimoncatalog.order_dw.dwd_orders
WHERE pay_id IS NOT NULL AND order_fee IS NOT NULL;

即時消費DWM層dwm_users_shops表的變更資料

在Realtime Compute控制台的資料開發 > ETL頁面，建立名為dws的SQL流作業，並將如下代碼複製到SQL編輯器後，部署作業並無狀態啟動作業。

通過該SQL作業，dwm_users_shops表的資料會寫入dws_users表和dws_shops表中，利用Paimon表的預彙總資料合併機制，在dws_users表中，計算每個使用者的總消費額（paid_buy_fee_sum），在dws_shops表中計算商戶的總流水（paid_buy_fee_sum），商戶的消費使用者數量（對1求和）和消費總人次（pv）。

SET 'execution.checkpointing.max-concurrent-checkpoints' = '3';
SET 'table.exec.sink.upsert-materialize' = 'NONE';
SET 'execution.checkpointing.interval' = '10s';
SET 'execution.checkpointing.min-pause' = '10s';
-- 與dwd不同，此處每一條INSERT語句寫入的是不同的Paimon表，可以放在同一個作業中。
BEGIN STATEMENT SET;
INSERT INTO paimoncatalog.order_dw.dws_users
SELECT 
    user_id,
    ds,
    paid_buy_fee_sum
FROM paimoncatalog.order_dw.dwm_users_shops;
-- 以商戶為主鍵，部分熱門商戶的資料量可能遠高於其他商戶。
-- 因此使用local merge在寫入Paimon之前先在記憶體中進行預彙總，緩解資料扭曲問題。
INSERT INTO paimoncatalog.order_dw.dws_shops /*+ OPTIONS('local-merge-buffer-size' = '64mb') */
SELECT
    shop_id,
    ds,
    paid_buy_fee_sum,
    1, -- 一條輸入記錄代表一名使用者在該商戶的所有消費
    pv
FROM paimoncatalog.order_dw.dwm_users_shops;
END;

查看dws_users表和dws_shops表的資料

在Realtime Compute控制台的資料開發 > 資料查詢頁面的查詢指令碼頁簽，將如下代碼拷貝到查詢指令碼後，選中目標片段後單擊右上方的運行。
```
--查看dws_users表資料
SELECT * FROM paimoncatalog.order_dw.dws_users ORDER BY user_id;
```
執行查詢後，dws_users 表返回三條記錄，包含 user_id、ds、payed_buy_fee_sum 三列：user_001 對應 8,000.00，user_002 對應 5,000.00，user_003 對應 5,000.00，日期均為 20230215。
```
--查看dws_shops表資料
SELECT * FROM paimoncatalog.order_dw.dws_shops ORDER BY shop_id;
```
執行 dws_shops 表查詢後，結果包含 shop_id、ds、payed_buy_fee_sum、uv、pv 共 5 列，返回 4 行資料（shop_id 為 12345～12348，ds 均為 20230215，payed_buy_fee_sum 分別為 5000.00、4000.00、7000.00、2000.00，uv 分別為 1、1、2、2，pv 分別為 1、1、3、2）。

捕捉業務資料庫的變化

前面已完成了流式湖倉的構建，下面將測試流式湖倉捕捉業務資料庫變化的能力。

向MySQL的order_dw資料庫中插入如下資料。

INSERT INTO orders VALUES
(100008, 'user_001', 12345, 3, 3000, '2023-02-15 17:40:56', '2023-02-15 18:42:56', 1),
(100009, 'user_002', 12348, 4, 1000, '2023-02-15 18:40:56', '2023-02-15 19:42:56', 1),
(100010, 'user_003', 12348, 2, 2000, '2023-02-15 19:40:56', '2023-02-15 20:42:56', 1);
INSERT INTO orders_pay VALUES
(2008, 100008, 1, '2023-02-15 18:40:56'),
(2009, 100009, 1, '2023-02-15 19:40:56'),
(2010, 100010, 0, '2023-02-15 20:40:56');

查看dws_users表和dws_shops表的資料。在Realtime Compute控制台的資料開發 > 資料查詢頁面的查詢指令碼頁簽，將如下代碼拷貝到查詢指令碼後，選中目標片段後單擊右上方的運行。
- dws_users表
```
SELECT * FROM paimoncatalog.order_dw.dws_users ORDER BY user_id;
```
  查詢返回三行資料，包含 user_id、ds、payed_buy_fee_sum 三列：user_001 / 20230215 / 11,000.00，user_002 / 20230215 / 6,000.00，user_003 / 20230215 / 7,000.00。
- dws_shops表
```
SELECT * FROM paimoncatalog.order_dw.dws_shops ORDER BY shop_id;
```
  查詢結果包含 5 列：shop_id、ds、payed_buy_fee_sum、uv、pv，共 4 行資料。shop_id 分別為 12345、12346、12347、12348，ds 均為 20230215，payed_buy_fee_sum 分別為 8000.00、4000.00、7000.00、5000.00，uv 分別為 1、1、2、3，pv 分別為 2、1、3、4。

使用流式湖倉

上一小節展示了在Flink中進行Paimon Catalog的建立與Paimon表的寫入。本節展示流式湖倉搭建完成後，利用StarRocks進行資料分析的一些簡單應用情境。

串連StarRocks與DLF

詳情請參見Serverless StarRocks訪問DLF。

排名查詢

對DWS層彙總表進行分析。本文使用StarRocks查詢23年2月15日交易額前三高的商戶的程式碼範例如下。

SELECT ROW_NUMBER() OVER (ORDER BY paid_buy_fee_sum DESC) AS rn, shop_id, paid_buy_fee_sum
FROM paimoncatalog.order_dw.dws_shops
WHERE ds = '20230215'
ORDER BY rn LIMIT 3;

查詢結果返回 3 行資料：rn=1 對應 shop_id=12345、payed_buy_fee_sum=8000.00；rn=2 對應 shop_id=12347、payed_buy_fee_sum=7000.00；rn=3 對應 shop_id=12348、payed_buy_fee_sum=5000.00。

明細查詢

對DWD層寬表進行分析。本文使用StarRocks查詢某個客戶23年2月特定支付平台支付的訂單明細的程式碼範例如下。

SELECT * FROM paimoncatalog.order_dw.dwd_orders
WHERE order_create_time >= '2023-02-01 00:00:00' AND order_create_time < '2023-03-01 00:00:00'
AND order_user_id = 'user_001'
AND pay_platform = 0
ORDER BY order_create_time;;

查詢返回結果如下。

   order_id  order_user_id  order_shop_id  order_product_id  order_product_catalog_name  order_fee
0  100006    user_001       12348          1                 phone_aaa                   1000
1  100004    user_001       12347          4                 phone_ddd                   2000

資料報表

對DWD層寬表進行分析。本文使用StarRocks查詢23年2月內每個品類的訂單總量和訂單總金額的程式碼範例如下。

SELECT
  order_create_time AS order_create_date,
  order_product_catalog_name,
  COUNT(*),
  SUM(order_fee)
FROM
  paimoncatalog.order_dw.dwd_orders
WHERE
  order_create_time >= '2023-02-01 00:00:00'  and order_create_time < '2023-03-01 00:00:00'
GROUP BY
  order_create_date, order_product_catalog_name
ORDER BY
  order_create_date, order_product_catalog_name;

執行上述查詢後返回9條記錄，顯示2023-02-15 10:40:56至18:40:56期間按小時粒度的訂單資料。產品目錄包括phone_ddd、phone_aaa、phone_eee、phone_ccc、phone_bbb等，每條記錄的count(*)均為1，sum(order_fee)值分別為2000、1000、1000、2000、3000、4000、5000、3000、1000。

Realtime Compute for Apache Flink：Paimon+StarRocks流式湖倉構建