ToB企服应用市场:ToB评测及商务社交产业平台

标题: Flink-StarRocks详解:第六部门-即席查询大案例解析(第56天) [打印本页]

作者: 种地 时间: 2024-8-9 14:13
标题: Flink-StarRocks详解:第六部门-即席查询大案例解析(第56天)
系列文章目次

数仓场景：即席查询案例
6.1 场景介绍
6.2 方案架构
6.3 方案特点
6.4 利用流程
6.4.1 步骤一：创建MySQL源数据表
6.4.2 步骤二：创建StarRocks表
6.4.3 步骤三：执行Flink任务，启动数据流
6.4.4 步骤四：验证数据

前言

本文为Flink-StarRocks详解后续章节：主要详解StarRocks数仓场景：即席查询大案例
6. 数仓场景：即席查询案例

本文通过示例介绍如何基于EMR Serverless StarRocks的视图能力构建数仓场景-即席查询办理方案。
6.1 场景介绍

随着向量化、CBO（Cost Based Optimizer，基于代价的优化器）、单机多核调理等技术的应用，StarRocks的盘算能力逐步提升。许多时候在使用StarRocks举行数仓分层建模时，大部门将数据建模到DWD层（底子整合层）或DWS层（维度宽度）。在实际业务中，运用StarRocks的盘算能力，可以直接查询DWD或DWS层数据，还可以机动地交互式即席查询。
6.2 方案架构

使用StarRocks实现数仓场景即席查询的基本架构如下图所示。

整体数据流如下：
（1）Flink洗濯导入Kafka的日志大概通过Flink-CDC-StarRocks工具读取MySQL Binlog导入StarRocks。根据需要选用明细、聚合、更新或主键各种模子，只物理落地ODS层（格式整理层）。
（2）向上接纳StarRocks View视图能力，利用StarRocks向量化极速查询和CBO优化器满足多表关联、嵌套子查询等复杂SQL，查询时现场盘算指标结果，保证指标上卷和下钻高度同源同等。
6.3 方案特点

该方案主要特点是，盘算逻辑在StarRocks侧（现场查询），实用于业务库高频数据更新的场景，实体数据只在ODS或DWD层存储。
 方案优势
 机动性强，可随时根据业务逻辑调整View。
 指标修正简单，上层都是View逻辑封装，只需要更新底表数据。
 方案缺点
当View的逻辑较为复杂，数据量较多时，查询性能较低。
 实用场景
 数据泉源于数据库和埋点系统，适合对QPS要求不高，对机动性要求比力高，且盘算资源较为充足的场景。
 实时要求非常高，要求写入即可查，更新即反馈。适合有即席查询需求，且资源较为充足，查询复杂度较低的场景。
6.4 利用流程

6.4.1 步骤一：创建MySQL源数据表

（1）登录DMS
（2）创建库和表

CREATE DATABASE IF NOT EXISTS flink_cdc;
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 )
);
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 )
);

复制代码

6.4.2 步骤二：创建StarRocks表

（1）登录EMR StarRocks Manager
（2）创建库和表

CREATE DATABASE IF NOT EXISTS `flink_cdc`;
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"
);
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"
);

复制代码

（3）基于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;

复制代码

（4）基于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;

复制代码

6.4.3 步骤三：执行Flink任务，启动数据流

（1）打开阿里云flink控制台
（2）创建MySQL CDC映射表
注意：hostname等需要根据自己的实际情况举行修改。

CREATE DATABASE IF NOT EXISTS `vvp`.`flinkcdc`;
CREATE TABLE IF NOT EXISTS `vvp`.`flinkcdc`.`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',
'hostname' = 'rm-cn-x0r3fp1lj000qa.rwlb.rds.aliyuncs.com',
'port' = '3306',
'username' = 'xxxxxx',
'password' = 'xxxxxx',
'database-name' = 'flink_cdc',
'table-name' = 'customers'
);
CREATE TABLE IF NOT EXISTS `vvp`.`flinkcdc`.`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 (
'connector' = 'mysql',
'hostname' = 'rm-cn-x0r3fp1lj000qa.rwlb.rds.aliyuncs.com',
'port' = '3306',
'username' = 'xxxx',
'password' = 'xxxxxx!',
'database-name' = 'flink_cdc',
'table-name' = 'orders'
);

复制代码

（3）创建StarRocks映射表
注意：jdbc-url、load-url等需要根据自己的实际情况举行修改。查询位置为EMR控制台-》StarRocks-》点击实例-》实例详情

CREATE TABLE IF NOT EXISTS `vvp`.`flinkcdc`.`customers_sink` (
`customer_id` INT NOT NULL,
`customer_age` FLOAT NOT NULL,
`customer_name` STRING NOT NULL,
PRIMARY KEY(`customer_id`)
NOT ENFORCED
) with (
'connector' = 'starrocks'
,'jdbc-url' = 'jdbc:mysql://fe-c-838fd7a4db1550a6-internal.starrocks.aliyuncs.com:9030'
,'load-url' = 'fe-c-838fd7a4db1550a6-internal.starrocks.aliyuncs.com:8030'
,'database-name' = 'flink_cdc'
,'table-name' = 'customers'
,'username' = 'xxxxxx'
,'password' = 'xxxxxx'
,'sink.buffer-flush.interval-ms' = '5000'
,'sink.semantic' = 'exactly-once'
);
CREATE TABLE IF NOT EXISTS `vvp`.`flinkcdc`.`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 (
'connector' = 'starrocks'
,'jdbc-url' = 'jdbc:mysql://fe-c-838fd7a4db1550a6-internal.starrocks.aliyuncs.com:9030'
,'load-url' = 'fe-c-838fd7a4db1550a6-internal.starrocks.aliyuncs.com:8030'
,'database-name' = 'flink_cdc'
,'table-name' = 'orders'
,'username' = 'xxxxxx''
,'password' = 'xxxxxx'
,'sink.buffer-flush.interval-ms' = '5000'
,'sink.semantic' = 'exactly-once'
);

复制代码

 参数含义

（4）将MySQL数据插入到StarRocks
以下代码写到一个流作业中，然后摆设运行。
需要无状态启动，并且设置checkpoint周期为5秒

BEGIN STATEMENT SET;
INSERT INTO `vvp`.`flinkcdc`.`customers_sink` SELECT * FROM `vvp`.`flinkcdc`.`customers_src`;
INSERT INTO `vvp`.`flinkcdc`.`orders_sink` SELECT * FROM `vvp`.`flinkcdc`.`orders_src`;
END;

复制代码

6.4.4 步骤四：验证数据

（1）在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");

复制代码

（2）在EMR StarRocks Manager中举行查询
1）查看orders表信息