本文通過示例為您介紹如何基于EMR Serverless StarRocks的視圖能力構建數倉場景-即席查詢解決方案。
前提條件
已創建DataFlow或自定義集群，具體操作請參見創建集群。
已創建EMR Serverless StarRocks實例，具體操作請參見創建實例。
已創建RDS MySQL，具體操作請參見快速創建RDS MySQL實例。
說明本文示例中DataFlow集群為EMR-3.42.0版本、MySQL為5.7版本。
使用限制
DataFlow集群、EMR Serverless StarRocks實例和RDS MySQL實例需要在同一個VPC下，并且在同一個可用區下。
DataFlow集群和EMR Serverless StarRocks實例均須開啟公網訪問。
RDS MySQL為5.7及以上版本。
注意事項
本文檔僅供測試使用，生產級別的Flink作業請使用阿里云實時計算Flink版產品進行配置，或者使用YARN或者Kubernetes提交作業。
場景介紹
隨著向量化、CBO（Cost Based Optimizer，基于代價的優化器）、單機多核調度等技術的應用，StarRocks的計算能力逐步提升。很多時候您在使用StarRocks進行數倉分層建模時，大部分將數據建模到DWD層（基礎整合層）或DWS層（維度寬度）。在實際業務中，運用StarRocks的計算能力，可以直接查詢DWD或DWS層數據，還可以靈活地交互式即席查詢。
方案架構
使用StarRocks實現數倉場景即席查詢的基本架構如下圖所示。
整體數據流如下：
Flink清洗導入Kafka的日志或者通過Flink-CDC-StarRocks工具讀取MySQL Binlog導入StarRocks。根據需要選用明細、聚合、更新或主鍵各種模型，只物理落地ODS層（格式整理層）。
向上采用StarRocks View視圖能力，利用StarRocks向量化極速查詢和CBO優化器滿足多表關聯、嵌套子查詢等復雜SQL，查詢時現場計算指標結果，保證指標上卷和下鉆高度同源一致。
方案特點
該方案主要特點是，計算邏輯在StarRocks側（現場查詢），適用于業務庫高頻數據更新的場景，實體數據只在ODS或DWD層存儲。
方案優勢
靈活性強，可隨時根據業務邏輯調整View。
指標修改簡單，上層都是View邏輯封裝，只需要更新底表數據。
方案缺點
當View的邏輯較為復雜，數據量較多時，查詢性能較低。
適用場景
數據來源于數據庫和埋點系統，適合對QPS要求不高，對靈活性要求比較高，且計算資源較為充足的場景。
實時要求非常高，要求寫入即可查，更新即反饋。適合有即席查詢需求，且資源較為充足，查詢復雜度較低的場景。
操作流程
示例操作如下：
步驟一：創建MySQL源數據表
創建測試的數據庫和賬號，具體操作請參見創建數據庫和賬號。
創建完數據庫和賬號后，需要授權測試賬號的讀寫權限。
說明本文示例中創建的數據庫名稱為flink_cdc，賬號為emr_test。
使用創建的測試賬號連接MySQL實例，具體操作請參見通過DMS登錄RDS MySQL。
執行以下命令，創建數據表orders。
create table flink_cdc.orders ( order_id INT NOT NULL AUTO_INCREMENT, order_revenue FLOAT NOT NULL, order_region VARCHAR(40) NOT NULL, customer_id INT NOT NULL, PRIMARY KEY (order_id) );執行以下命令，創建數據表customers。
create table flink_cdc.customers ( customer_id INT NOT NULL, customer_age INT NOT NULL, customer_name VARCHAR(40) NOT NULL, PRIMARY KEY (customer_id) );
步驟二：創建StarRocks表
連接EMR Serverless StarRocks實例，詳情請參見通過客戶端方式連接StarRocks實例。
執行以下命令，創建數據庫。
CREATE DATABASE IF NOT EXISTS `flink_cdc`;執行以下命令，創建數據表customers。
CREATE TABLE IF NOT EXISTS `flink_cdc`.`customers` ( `customer_id` INT NOT NULL COMMENT "", `customer_age` FLOAT NOT NULL COMMENT "", `customer_name` STRING NOT NULL COMMENT "" ) ENGINE=olap PRIMARY KEY(`customer_id`) COMMENT "" DISTRIBUTED BY HASH(`customer_id`) BUCKETS 1 PROPERTIES ( "replication_num" = "1" );執行以下命令，創建數據表orders。
CREATE TABLE IF NOT EXISTS `flink_cdc`.`orders` ( `order_id` INT NOT NULL COMMENT "", `order_revenue` FLOAT NOT NULL COMMENT "", `order_region` STRING NOT NULL COMMENT "", `customer_id` INT NOT NULL COMMENT "" ) ENGINE=olap PRIMARY KEY(`order_id`) COMMENT "" DISTRIBUTED BY HASH(`order_id`) BUCKETS 1 PROPERTIES ( "replication_num" = "1" );執行以下命令，基于ODS表創建DWD視圖。
CREATE VIEW flink_cdc.dwd_order_customer_valid ( order_id, order_revenue, order_region, customer_id, customer_age, customer_name ) AS SELECT o.order_id, o.order_revenue, o.order_region, c.customer_id, c.customer_age, c.customer_name FROM flink_cdc.customers c JOIN flink_cdc.orders o ON c.customer_id=o.customer_id WHERE c.customer_id != -1;執行以下命令，基于DWD表創建DWS視圖。
CREATE VIEW flink_cdc.dws_agg_by_region ( order_region, order_cnt, order_total_revenue) AS SELECT order_region, count(order_region), sum(order_revenue) FROM flink_cdc.dwd_order_customer_valid GROUP BY order_region;
步驟三：執行Flink任務，啟動數據流
下載Flink CDC connector和Flink StarRocks Connector，并上傳到DataFlow集群的/opt/apps/FLINK/flink-current/lib目錄下。
使用SSH方式登錄DataFlow集群，詳情請參見登錄集群。
添加端口配置，并修改并行執行的任務槽數量。
執行以下命令，編輯文件flink-conf.yaml。
vim /etc/taihao-apps/flink-conf/flink-conf.yaml添加以下內容至文件最后一行。
rest.port: 8083修改參數
taskmanager.numberOfTaskSlots的值為3，默認值為1。
執行以下命令，啟動集群。
重要本文示例僅供測試，如果是生產級別的Flink作業請使用YARN或Kubernetes方式提交，詳情請參見Apache Hadoop YARN和Native Kubernetes。
/opt/apps/FLINK/flink-current/bin/start-cluster.sh編寫Flink SQL作業，并保存為demo.sql。
執行以下命令，編輯demo.sql文件。
vim demo.sql文件內容如下所示。
CREATE DATABASE IF NOT EXISTS `default_catalog`.`flink_cdc`; CREATE TABLE IF NOT EXISTS `default_catalog`.`flink_cdc`.`customers_src` ( `customer_id` INT NOT NULL, `customer_age` FLOAT NOT NULL, `customer_name` STRING NOT NULL, PRIMARY KEY(`customer_id`) NOT ENFORCED ) with ( 'connector' = 'mysql-cdc', 'hostname' = 'rm-2ze8398257383****.mysql.rds.aliyuncs.com', 'port' = '3306', 'username' = 'emr_test', 'password' = 'Yz12****', 'database-name' = 'flink_cdc', 'table-name' = 'customers' ); CREATE TABLE IF NOT EXISTS `default_catalog`.`flink_cdc`.`customers_sink` ( `customer_id` INT NOT NULL, `customer_age` FLOAT NOT NULL, `customer_name` STRING NOT NULL, PRIMARY KEY(`customer_id`) NOT ENFORCED ) with ( 'load-url' = 'fe-c-9b354c83e891****-internal.starrocks.aliyuncs.com:8030', 'database-name' = 'flink_cdc', 'jdbc-url' = 'jdbc:mysql://fe-c-9b354c83e891****-internal.starrocks.aliyuncs.com:9030', 'sink.buffer-flush.interval-ms' = '15000', 'sink.properties.format' = 'json', 'username' = 'admin', 'table-name' = 'customers', 'sink.properties.strip_outer_array' = 'true', 'password' = '1qaz!QAZ', 'sink.max-retries' = '10', 'connector' = 'starrocks' ); INSERT INTO `default_catalog`.`flink_cdc`.`customers_sink` SELECT * FROM `default_catalog`.`flink_cdc`.`customers_src`; CREATE TABLE IF NOT EXISTS `default_catalog`.`flink_cdc`.`orders_src` ( `order_id` INT NOT NULL, `order_revenue` FLOAT NOT NULL, `order_region` STRING NOT NULL, `customer_id` INT NOT NULL, PRIMARY KEY(`order_id`) NOT ENFORCED ) with ( 'database-name' = 'flink_cdc', 'table-name' = 'orders', 'connector' = 'mysql-cdc', 'hostname' = 'rm-2ze8398257383****.mysql.rds.aliyuncs.com', 'port' = '3306', 'username' = 'emr_test', 'password' = 'Yz12****' ); CREATE TABLE IF NOT EXISTS `default_catalog`.`flink_cdc`.`orders_sink` ( `order_id` INT NOT NULL, `order_revenue` FLOAT NOT NULL, `order_region` STRING NOT NULL, `customer_id` INT NOT NULL, PRIMARY KEY(`order_id`) NOT ENFORCED ) with ( 'sink.properties.strip_outer_array' = 'true', 'password' = '1qaz!QAZ', 'sink.max-retries' = '10', 'connector' = 'starrocks', 'table-name' = 'orders', 'jdbc-url' = 'jdbc:mysql://fe-c-9b354c83e891****-internal.starrocks.aliyuncs.com:9030', 'sink.buffer-flush.interval-ms' = '15000', 'sink.properties.format' = 'json', 'username' = 'admin', 'load-url' = 'fe-c-9b354c83e891****-internal.starrocks.aliyuncs.com:8030', 'database-name' = 'flink_cdc' ); INSERT INTO `default_catalog`.`flink_cdc`.`orders_sink` SELECT * FROM `default_catalog`.`flink_cdc`.`orders_src`;涉及參數如下所示：
創建數據表customers_src。
參數
描述
connector
固定值為mysql-cdc。
hostname
RDS的內網地址。
您可以在RDS的數據庫連接頁面，單擊內網地址進行復制。例如，rm-2ze8398257383****.mysql.rds.aliyuncs.com。
port
固定值為3306。
username
步驟一：創建MySQL源數據表中創建的賬號名。本示例為emr_test。
password
步驟一：創建MySQL源數據表中創建的賬號的密碼。本示例為Yz12****。
database-name
步驟一：創建MySQL源數據表中創建的數據庫名。本示例為flink_cdc。
table-name
步驟一：創建MySQL源數據表中創建的數據表。本示例為customers。
創建數據表customers_sink和orders_sink。
參數
描述
load-url
指定FE節點的內網地址和HTTP端口，格式為
EMR Serverless StarRocks實例FE節點的內網地址:8030。例如，fe-c-9b354c83e891****-internal.starrocks.aliyuncs.com:8030。說明關于如何獲取EMR Serverless StarRocks實例FE節點的內網地址，請參見查看實例列表與詳情。
database-name
步驟一：創建MySQL源數據表中創建的數據庫名。本示例為flink_cdc。
jdbc-url
用于在StarRocks中執行查詢操作。
例如，jdbc:mysql://fe-c-9b354c83e891****-internal.starrocks.aliyuncs.com:9030。其中，
fe-c-9b354c83e891****-internal.starrocks.aliyuncs.com為EMR Serverless StarRocks實例FE節點的內網地址。說明關于如何獲取EMR Serverless StarRocks實例FE節點的內網地址，請參見查看實例列表與詳情。
username
StarRocks連接用戶名。固定為admin。
table-name
本示例固定值為customers。
connector
固定值為starrocks。
執行以下命令，啟動Flink任務。
/opt/apps/FLINK/flink-current/bin/sql-client.sh -f demo.sql
步驟四：驗證數據
使用步驟一：創建MySQL源數據表中創建的測試賬號連接MySQL實例，具體操作請參見通過DMS登錄RDS MySQL。
在RDS數據庫窗口執行以下命令，向表orders和customers中插入數據。
INSERT INTO flink_cdc.orders(order_id,order_revenue,order_region,customer_id) VALUES(1,10,"beijing",1); INSERT INTO flink_cdc.orders(order_id,order_revenue,order_region,customer_id) VALUES(2,10,"beijing",1); INSERT INTO flink_cdc.customers(customer_id,customer_age,customer_name) VALUES(1, 22, "emr_test");連接EMR Serverless StarRocks實例，詳情請參見通過客戶端方式連接StarRocks實例。
執行以下命令，查詢ODS層數據。
執行以下命令，查看orders表信息。
select * from flink_cdc.orders;返回信息如下所示。
+----------+---------------+--------------+-------------+ | order_id | order_revenue | order_region | customer_id | +----------+---------------+--------------+-------------+ | 1 | 10 | beijing | 1 | | 2 | 10 | beijing | 1 | +----------+---------------+--------------+-------------+執行以下命令，查看customers表信息。
select * from flink_cdc.customers;返回信息如下所示。
+-------------+--------------+---------------+ | customer_id | customer_age | customer_name | +-------------+--------------+---------------+ | 1 | 22 | emr_test | +-------------+--------------+---------------+
執行以下命令，查詢DWD層數據。
select * from flink_cdc.dwd_order_customer_valid;返回信息如下所示。
+----------+---------------+--------------+-------------+--------------+---------------+ | order_id | order_revenue | order_region | customer_id | customer_age | customer_name | +----------+---------------+--------------+-------------+--------------+---------------+ | 1 | 10 | beijing | 1 | 22 | emr_test | | 2 | 10 | beijing | 1 | 22 | emr_test | +----------+---------------+--------------+-------------+--------------+---------------+ 2 rows in set (0.00 sec)執行以下命令，查詢DWS層數據。
select * from flink_cdc.dws_agg_by_region;返回信息如下所示。
+--------------+-----------+---------------------+ | order_region | order_cnt | order_total_revenue | +--------------+-----------+---------------------+ | beijing | 2 | 20 | +--------------+-----------+---------------------+ 1 row in set (0.01 sec)