全部產品
Search

搜尋本產品

    E-MapReduce：數倉情境：即席查詢

    文件中心

    E-MapReduce：數倉情境：即席查詢

    更新時間：Jul 10, 2024

    本文通過樣本為您介紹如何基於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作業請使用阿里雲Realtime ComputeFlink版產品進行配置，或者使用YARN或者Kubernetes提交作業。

    詳情請參見Apache Hadoop YARN和Native Kubernetes。

    情境介紹

    隨著向量化、CBO（Cost Based Optimizer，基於代價的最佳化器）、單機多核調度等技術的應用，StarRocks的計算能力逐步提升。很多時候您在使用StarRocks進行數倉分層建模時，大部分將資料建模到DWD層（基礎整合層）或DWS層（維度寬度）。在實際業務中，運用StarRocks的計算能力，可以直接查詢DWD或DWS層資料，還可以靈活地互動式即席查詢。

    方案架構

    整體資料流如下：

    1. Flink清洗匯入Kafka的日誌或者通過Flink-CDC-StarRocks工具讀取MySQL Binlog匯入StarRocks。根據需要選用明細、彙總、更新或主鍵各種模型，只物理落地ODS層（格式整理層）。

    2. 向上採用StarRocks View視圖能力，利用StarRocks向量化極速查詢和CBO最佳化器滿足多表關聯、嵌套子查詢等複雜SQL，查詢時現場計算指標結果，保證指標上卷和下鑽高度同源一致。

    方案特點

    該方案主要特點是，計算邏輯在StarRocks側（現場查詢），適用於業務庫高頻資料更新的情境，實體資料只在ODS或DWD層儲存。

    • 方案優勢

      • 靈活性強，可隨時根據商務邏輯調整View。

      • 指標修改簡單，上層都是View邏輯封裝，只需要更新底表資料。

    • 方案缺點

      當View的邏輯較為複雜，資料量較多時，查詢效能較低。

    • 適用情境

      • 資料來源於資料庫和埋點系統，適合對QPS要求不高，對靈活性要求比較高，且計算資源較為充足的情境。

      • 即時要求非常高，要求寫入即可查，更新即反饋。適合有即席查詢需求，且資源較為充足，查詢複雜度較低的情境。

    操作流程

    樣本操作如下：

    1. 步驟一：建立MySQL來源資料表

    2. 步驟二：建立StarRocks表

    3. 步驟三：執行Flink任務，啟動資料流

    4. 步驟四：驗證資料

    步驟一：建立MySQL來源資料表

    1. 建立測試的資料庫和帳號，具體操作請參見建立資料庫和帳號。

      建立完資料庫和帳號後，需要授權測試帳號的讀寫權限。

      說明

      本文樣本中建立的資料庫名稱為flink_cdc，帳號為emr_test。

    2. 使用建立的測試帳號串連MySQL執行個體，具體操作請參見通過DMS登入RDS MySQL。

    3. 執行以下命令，建立資料表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)
);

    4. 執行以下命令，建立資料表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表

    1. 串連EMR Serverless StarRocks執行個體，詳情請參見通過用戶端方式串連StarRocks執行個體。

    2. 執行以下命令，建立資料庫。

      CREATE DATABASE IF NOT EXISTS `flink_cdc`;

    3. 執行以下命令，建立資料表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"
);

    4. 執行以下命令，建立資料表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"
);

    5. 執行以下命令，基於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;

    6. 執行以下命令，基於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任務，啟動資料流

    1. 下載Flink CDC connector和Flink StarRocks Connector，並上傳到DataFlow叢集的/opt/apps/FLINK/flink-current/lib目錄下。

    2. 使用SSH方式登入DataFlow叢集，詳情請參見登入叢集。

    3. 添加連接埠配置，並修改並存執行的任務槽數量。

      1. 執行以下命令，編輯檔案flink-conf.yaml。

        vim /etc/taihao-apps/flink-conf/flink-conf.yaml

      2. 添加以下內容至檔案最後一行。

        rest.port: 8083

      3. 修改參數taskmanager.numberOfTaskSlots的值為3，預設值為1。

    4. 執行以下命令，啟動叢集。

      重要

      本文樣本僅供測試，如果是生產層級的Flink作業請使用YARN或Kubernetes方式提交，詳情請參見Apache Hadoop YARN和Native Kubernetes。

      /opt/apps/FLINK/flink-current/bin/start-cluster.sh

    5. 編寫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。

    6. 執行以下命令，啟動Flink任務。

       /opt/apps/FLINK/flink-current/bin/sql-client.sh -f demo.sql

    步驟四：驗證資料

    1. 使用步驟一：建立MySQL來源資料表中建立的測試帳號串連MySQL執行個體，具體操作請參見通過DMS登入RDS MySQL。

    2. 在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");

    3. 串連EMR Serverless StarRocks執行個體，詳情請參見通過用戶端方式串連StarRocks執行個體。

    4. 執行以下命令，查詢ODS層資料。

      1. 執行以下命令，查看orders表資訊。

        select * from flink_cdc.orders;

        返回資訊如下所示。

        +----------+---------------+--------------+-------------+
| order_id | order_revenue | order_region | customer_id |
+----------+---------------+--------------+-------------+
|        1 |            10 | beijing      |           1 |
|        2 |            10 | beijing      |           1 |
+----------+---------------+--------------+-------------+

      2. 執行以下命令，查看customers表資訊。

        select * from flink_cdc.customers;

        返回資訊如下所示。

        +-------------+--------------+---------------+
| customer_id | customer_age | customer_name |
+-------------+--------------+---------------+
|           1 |           22 | emr_test      |
+-------------+--------------+---------------+

    5. 執行以下命令，查詢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)

    6. 執行以下命令，查詢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)