揭秘“湖仓一体”——Flink+Paimon+StarRocks，打造实时分析新纪元

1.湖仓一体

数据湖仓是 Flink 流批一体发挥重要作用的场景,使用 Flink + Paimon + starRocks 来构建湖仓一体数据分析.
Apache Paimon 是一个专为实时数据处理而设计的湖表格式，它最大的亮点是使用了 LSM Tree 技术。与 Hudi 相比，Paimon 在更新插入（Upsert）操作上速度快了4倍，查询扫描（Scan）速度也提高了10倍。这意味着它能提供更快的响应速度，同时降低数据入湖的成本，并且让开发者用起来更高效。Paimon 社区十分活跃，很多产品都在迅速与其兼容，这让它的生态系统发展得比其他湖库表格式更快、更全面。
StarRocks 是一款高性能分析型数据仓库，使用向量化、MPP 架构、CBO、智能物化视图、可实时更新的列式存储引擎等技术实现多维、实时、高并发的数据分析。StarRocks 既支持从各类实时和离线的数据源高效导入数据，也支持直接分析数据湖上各种格式的数据。StarRocks 兼容 MySQL 协议，可使用 MySQL 客户端和常用 BI 工具对接。同时 StarRocks 具备水平扩展，高可用、高可靠、易运维等特性。广泛应用于实时数仓、OLAP 报表、数据湖分析等场景。
湖仓一体化.jpg

Flink + Paimon + StarRocks 流式湖仓方案将 3 个产品做了非常紧密的结合，首先使用 Flink 流批一体计算引擎将数仓以 Paimon 格式在湖上构建，使用 Flink 完成数仓 ODS 到 DWD 层，DWS 和 ADS 的计算。通过使用 StarRocks 对各层数仓做统一的 OLAP 查询和 ADS 层在线分析。基于 Paimon 可以实现高吞吐入湖；基于 Flink 可以实现全链路的流批一体计算，基于 StarRocks 可以实现高性能的 OLAP 查询，所以整个链路从实时性、时效性、成本几个方面都可以取得比较好的平衡。
使用 StarRocks 统一管理数据湖和数据仓库，将高并发和实时性要求很高的业务放在 StarRocks 中分析，也可以使用 External Catalog 和外部表进行数据湖上的分析。StarRocks 从 3.1 版本开始支持 Paimon Catalog。
Paimon Catalog 是一种 External Catalog。通过 Paimon Catalog，您不需要执行数据导入就可以直接查询 Apache Paimon 里的数据。
在数据湖仓场景下，使用 Flink 可以完成复杂的数据拼接以及聚合计算，并且达到很高的实时性的要求。另外，实时链路在使用的过程中不可避免的会因为一些数据延迟等问题导致会有数据修正和数据回溯的需求。Flink 流批一体的特性能够让用户方便的使用与实时链路一样的作业代码，高效地完成数据修正和数据回溯的需求。

2.演示架构

通过flink-cdc 监听MySQL Binlog数据同步到Paimon ODS层,然后进行DWD数据清洗宽表打通,再到DWS层进行多维度汇总聚合,最后同ADS层进行数据呈现.其中用到streamPark进行作业编排.
2.1 组件使用版本

  • flink1.18.1
  • paimon0.8
  • fink-cdc3.1
  • streamPark2.1.4
  • starRocks3.1
    安装方式请自行安装

2.2 场景说明
在mysql创建3张表:用户表users,订单表orders,商品表products,订单详情表order_details
分析3个指标: 用户的总购买金额 产品的销售数量 用户的平均订单金额
在执行下面操作前,确保mysql,flink,streamPark,starRocks已经启动.

步骤一:准备演示数据

  1. 在MySQL中执行以下命令，创建数据表。
use emp;

-- 用户表
CREATE TABLE users (
    user_id BIGINT AUTO_INCREMENT PRIMARY KEY COMMENT '用户ID，主键',
    user_name VARCHAR(50) NOT NULL COMMENT '用户名',
    email VARCHAR(100) NOT NULL COMMENT '邮箱',
    registration_date DATE NOT NULL COMMENT '注册日期',
    PRIMARY KEY (`user_id`) USING BTREE
) COMMENT '用户表';

INSERT INTO users (user_id, user_name, email, registration_date) VALUES (1, '张山', 'alice@example.com', '2023-01-15');
INSERT INTO users (user_id, user_name, email, registration_date) VALUES (2, '李四', 'bob@example.com', '2023-02-20');
INSERT INTO users (user_id, user_name, email, registration_date) VALUES (3, '刘博', 'charlie@example.com', '2023-03-10');

-- 订单表
CREATE TABLE orders (
    order_id bigint AUTO_INCREMENT PRIMARY KEY COMMENT '订单ID，主键',
    user_id INT NOT NULL COMMENT '用户ID，外键，关联到users表',
    order_date DATE NOT NULL COMMENT '订单日期',
    total_amount DECIMAL(10, 2) NOT NULL COMMENT '订单总金额',
	PRIMARY KEY (`order_id`) USING BTREE
) COMMENT '订单表';

-- 演示数据
INSERT INTO orders (user_id, order_date, total_amount) VALUES
(1, '2023-04-01', 150.00),
(2, '2023-04-05', 200.00),
(3, '2023-04-10', 250.00),
(1, '2023-04-15', 300.00);

-- 商品表
CREATE TABLE products (
    product_id bigint AUTO_INCREMENT PRIMARY KEY COMMENT '产品ID，主键',
    product_name VARCHAR(100) NOT NULL COMMENT '产品名',
    price DECIMAL(10, 2) NOT NULL COMMENT '产品价格',
    PRIMARY KEY (`product_id`) USING BTREE
) COMMENT '产品表';

-- 演示数据
INSERT INTO products (product_name, price) VALUES
('笔记本', 50.00),
('手表', 75.00),
('耳机', 100.00);

-- 订单详情表
CREATE TABLE order_details (
    order_detail_id bigint AUTO_INCREMENT PRIMARY KEY COMMENT '订单详情ID，主键',
    order_id INT NOT NULL COMMENT '订单ID，外键，关联到orders表',
    product_id INT NOT NULL COMMENT '产品ID，外键，关联到products表',
    quantity INT NOT NULL COMMENT '购买数量',
    subtotal DECIMAL(10, 2) NOT NULL COMMENT '小计金额（quantity * price）',
     PRIMARY KEY (`order_detail_id`) USING BTREE
) COMMENT '订单详情表';

-- 演示数据
INSERT INTO order_details (order_id, product_id, quantity, subtotal) VALUES
(1, 1, 2, 100.00),
(1, 2, 1, 50.00),
(2, 1, 1, 50.00),
(2, 3, 2, 150.00),
(3, 2, 2, 150.00),
(3, 3, 1, 100.00),
(4, 3, 3, 300.00);

步骤二:mysql数据同步paimon

确保mysql已经开启binlog

1.编写flink-cdc同步任务,在flink-cdc的创建job文件夹,然后在里面创建mysql-to-paimon.yml

source:
type: mysql
name: MySQL Source
hostname: 192.168.1.72
port: 3306
username: root
password: 123456
tables: emp.users,emp.products,emp.order_details,emp.orders
server-id: 5401-5404

sink:
type: paimon
name: Paimon Sink
catalog.properties.metastore: filesystem
catalog.properties.warehouse: /opt/software/paimon

pipeline:
name: MySQL to Paimon Pipeline
parallelism: 1
  1. 在 flink-cdc的lib文件夹下添加:
    flink-cdc-pipeline-connector-mysql-3.1.0.jar
    flink-cdc-pipeline-connector-paimon-3.1.0.jar
    mysql-connector-java-8.0.27.jar
  2. 在保证flink集群启动的情况下,进行启动flink-cdc
 ./bin/flink-cdc.sh job/mysql-to-paimom.yaml --jar lib/mysql-connector-java-8.0.27.jar
# 执行成功会出现jobid
Pipeline has been submitted to cluster.
Job ID: b68bfad5753ae600eeb1efed17d957ff
Job Description: MySQL to Paimon Pipeline

4.来到flink工作台进行查询任务
image.png

5.在服务器上查看同步文件信息

 cd /opt/software/paimon/
 cd emp.db/
 ls
 #显示已经同步过来
order_details  orders  products  users

已经完成ODS层数据同步.

步骤三: DWD数据清洗宽表打通

1.打开streamPark,进行开始编写flink sql
image.png

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';
-- 创建CATALOG
CREATE CATALOG paimon_catalog WITH (
'type' = 'paimon',
'warehouse' = 'file:/opt/software/paimon'
);

-- 切换CATALOG
USE CATALOG paimon_catalog;
create DATABASE IF NOT EXISTS emp;
-- 切换database
use emp;

-- 创建dwd_user_orders表
CREATE TABLE IF NOT EXISTS dwd_user_orders (
order_id bigint,
user_id bigint,
user_name STRING,
order_date date,
total_amount decimal,
PRIMARY KEY (order_id) NOT ENFORCED
);

-- 创建dwd_orders_products_details表
CREATE TABLE IF NOT EXISTS dwd_orders_products_details (
order_detail_id bigint,
order_id bigint,
product_id bigint,
product_name STRING,
price decimal,
quantity bigint,
subtotal decimal,
PRIMARY KEY (order_detail_id) NOT ENFORCED
);

  

INSERT INTO
dwd_user_orders
SELECT
o.order_id,o.user_id,u.user_name,o.order_date,o.total_amount
FROM orders o join users u ON o.user_id=u.user_id;
INSERT INTO
dwd_orders_products_details
SELECT
d.order_detail_id,d.order_id,d.product_id,p.product_name,p.price,d.quantity,d.subtotal
FROM order_details d join products p ON p.product_id=d.product_id;

发布启动任务
image.png

flink-web-ui查看任务
image.png

步骤三:进行维度分析

创建DWS层进行多维度汇总聚合,还是在streamPark编写DWS层任务

统计维度指标:

  • 用户的总购买金额
  • 产品的销售数量
  • 订单的平均金额
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';
-- 创建CATALOG
CREATE CATALOG paimon_catalog WITH (
'type' = 'paimon',
'warehouse' = 'file:/opt/software/paimon'
);

  

-- 切换CATALOG
USE CATALOG paimon_catalog;
create DATABASE IF NOT EXISTS emp;
-- 切换database
use emp;
-- 创建用户的总购买金额表
CREATE TABLE IF NOT EXISTS dws_user_total_amount (
user_id bigint,
user_name STRING,
total_spent decimal,
PRIMARY KEY (user_id) NOT ENFORCED
);

-- 创建产品的销售数量
CREATE TABLE IF NOT EXISTS dws_product_sales_quantity (
product_id bigint,
product_name STRING,
total_quantity BIGINT,
PRIMARY KEY (product_id) NOT ENFORCED
);

-- 创建订单的平均金额
CREATE TABLE IF NOT EXISTS dws_order_average_amount (
order_id bigint,
average_order_amount decimal,
PRIMARY KEY (order_id) NOT ENFORCED
);

-- 用户的总购买金额
INSERT INTO
dws_user_total_amount
SELECT user_id,user_name, sum(total_amount) AS total_spent
FROM dwd_user_orders
group by user_id,user_name;
-- 产品的销售数量
INSERT INTO
dws_product_sales_quantity
SELECT product_id,product_name,SUM(quantity) AS total_quantity
FROM dwd_orders_products_details
group by product_id,product_name;
-- 订单的平均金额
INSERT INTO
dws_order_average_amount
SELECT order_id,AVG(total_amount) AS average_order_amount
FROM dwd_user_orders
group by order_id;

发布启动任务
image.png

flink-web-ui查看任务
image.png

步骤四:ADS查看维度结果数据

这里要使用starRocks进行查询paimon catalog数据表.在starRock 中paimon catalog是一种外部catalog.可以直接进行查询数据.
保证starRock正常启动.安装方式参考:https://www.cnblogs.com/freeweb/p/18137023
DBeaver连接上starRocks
image.png

查询对应维度数据
image.png

-- 查询用户的总购买金额
SELECT * FROM paimon_catalog.emp.dws_user_total_amount;
-- 结果
user_id|user_name|total_spent|
-------+---------+-----------+
      1|张山       |        450|
      2|李四       |        200|
      3|刘博       |        250|
  
-- 查询产品的销售数量
SELECT * FROM paimon_catalog.emp.dws_product_sales_quantity;
-- 结果
product_id|product_name|total_quantity|
----------+------------+--------------+
         1|笔记本         |             3|
         2|手表          |             3|
         3|耳机          |             6|
-- 查询订单的平均金额
SELECT * FROM paimon_catalog.emp.dws_order_average_amount;
-- 结果
order_id|average_order_amount|
--------+--------------------+
       1|                 150|
       2|                 200|
       3|                 250|
       4|                 300|

步骤五: 演示数据实时更新

在mysql表进行修改数据查询维度表数据是否发生计算结果变更
添加一条人员信息,产品信息,订单信息,订单详情信息,看维度表数据是否发生变化

  1. 在mysql中添加下面数据
INSERT INTO users (user_id, user_name, email, registration_date) 
VALUES (4, '刘晓天', 'charlie@example.com', '2024-06-17');

INSERT INTO orders (order_id,user_id, order_date, total_amount) VALUES
(5,4, '2024-06-17', 1800000.00);

INSERT INTO products (product_id,product_name, price) VALUES
(4,'天启坦克', 1800000.00);

INSERT INTO order_details (order_id, product_id, quantity, subtotal) VALUES
(5, 4, 1, 1800000.00);

sleep 5s

  1. 查看维度分析结果,已经发生结果变化.

删除,修改mysql表同样会触发维度结果变化
image.png

image.png
image.png

本文来自互联网用户投稿，该文观点仅代表作者本人，不代表本站立场。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如若转载，请注明出处：http://www.mfbz.cn/a/717040.html

如若内容造成侵权/违法违规/事实不符，请联系我们进行投诉反馈qq邮箱809451989@qq.com，一经查实，立即删除！

相关文章

中国星网时空信息集团春招Offer面经

中国星网时空信息集团春招Offer面经

本文介绍2024届春招中，中国卫星网络集团有限公司下属中国时空信息集团有限公司中，业务助理岗位1场面试的基本情况、提问问题等。 2024年04月投递了中国卫星网络集团有限公司下属中国时空信息集团有限公司中的业务助理岗位，所属部门为运营中心…
阅读更多...
第9章 类

第9章 类

第9章 类 9.1 创建和使用类9.1.1 创建 Dog 类9.1.2 根据类创建实例 9.2 使用类和实例9.2.1 Car 类9.2.2 给属性指定默认值9.2.3 修改属性的值 9.3 继承9.3.1 子类的方法__init__()9.3.2 给子类定义属性和方法9.3.3 重写父类的方法9.3.4 将实例用作属性9.3.5 模拟实物 9.4 导入类…
阅读更多...
短URL服务设计

短URL服务设计

引言 在营销系统里，为了增加系统的活跃用户数，经常会有各种各样的营销活动。这类活动几乎都是为了充分利用存量用户的价值，促使他们分享产品或App以达到触达到更多用户的目的。又或者是出于营销目的，群发优惠券触达短信这种场景。…
阅读更多...
充电学习—3、Uevent机制和其在android层的实现

充电学习—3、Uevent机制和其在android层的实现

sysfs 是 Linux userspace 和 kernel 进行交互的一个媒介。通过 sysfs，userspace 可以主动去读写 kernel 的一些数据，同样的， kernel 也可以主动将一些“变化”告知给 userspace。也就是说，通过sysfs，userspace 和 ker…
阅读更多...
欣九康诊疗系统助力诊所向数字化转型

欣九康诊疗系统助力诊所向数字化转型

数字化已经成为各行各业转型的重点方向，而为了不被时代所淘汰，医疗机构也势必要紧跟潮流，本人作为门诊部的负责人深知医疗机构要想实现数字化转型那么拥有一款便捷实用的医疗平台是必不可少的，近几年，随着国家大力支持…
阅读更多...
Ubuntu 在线或离线安装docker

Ubuntu 在线或离线安装docker

查看自己的ubuntu版本 在终端中执行以下命令： lsb_release -a 终端中的复制粘贴： ctrl shift c ctrl shifr v 在线安装docker（不需要外网）: 命令行安装：Ubuntu Docker -- 从入门到实践 看完…
阅读更多...
Ollama：本地部署大模型 + LobeChat：聊天界面 = 自己的ChatGPT

Ollama：本地部署大模型 + LobeChat：聊天界面 = 自己的ChatGPT

本地部署大模型 在本地部署大模型有多种方式，其中Ollama方式是最简单的，但是其也有一定的局限性，比如大模型没有其支持的GGUF二进制格式，就无法使用Ollama方式部署。 GGUF旨在实现快速加载和保存大语言模型，并易于阅读…
阅读更多...
香港Web3时代：比特币可以成为「收益性资产」吗？

香港Web3时代：比特币可以成为「收益性资产」吗？

原文标题：《CAN BITCOIN BE A PRODUCTIVE ASSET?》撰文：Pascal Hgli编译：Chris，Techub News本文来源香港Web3媒体 Techub News 比特币正在经历一场大的变化，人们对其性质有不同的看法。有些人将其视为日常交易的货币…
阅读更多...
ANSYS EMC解决方案与经典案例

ANSYS EMC解决方案与经典案例

EMC问题非常复杂，各行各业都会涉及，例如航空、航天、船舶、汽车、火车、高科技、物联网、消费电子。要考虑EMC的对象很多，包含整个系统、设备、PCB、线缆、电源、芯片封装。而且技术领域覆盖广，涉及高频问题、低频问题&#xff1b…
阅读更多...
AI大模型系统从入门到精通，看这一篇就够了

AI大模型系统从入门到精通，看这一篇就够了

前言 2023 年，人工智能发展达到新的里程碑。自 GPT 系列和 LLaMA 系列等大规模语言模型及应用问世以来，AI 内部技术突飞猛进，能力迅速超越以往。这些“超级 AI 助手”看似便捷强大，但其背后复杂原理及潜在影响值得深入思考。 这些…
阅读更多...
充电学习—5、healthed 电池服务

充电学习—5、healthed 电池服务

1、healthed服务监听接收内核kernel的电池事件，然后上传数据给framware层的batterysevice，BatteryService计算电池的电量，显示，绘制动画等 android电池系统框架： 2、healthd服务入口：android/system/cor…
阅读更多...
本地安装nightingale监控分析服务并发布公网详细流程

本地安装nightingale监控分析服务并发布公网详细流程

文章目录 前言1. Linux 部署Nightingale2. 本地访问测试3. Linux 安装cpolar4. 配置Nightingale公网访问地址5. 公网远程访问Nightingale管理界面6. 固定Nightingale公网地址 前言 本文主要介绍如何在本地Linux系统部署 Nightingale 夜莺监控并结合cpolar内网穿透工具实现远程…
阅读更多...
怎么把两个音频合成一个？将两个音频合成一个的四种方法

怎么把两个音频合成一个？将两个音频合成一个的四种方法

怎么把两个音频合成一个？在当今数字化的时代，音频处理已经成为我们生活中不可或缺的一部分。有时候，我们会希望将两段音频合成为一个，无论是为了制作音乐混音、创作声音效果，还是为了编辑播客节目或视频配音。合成音频…
阅读更多...
Qt第三方库QHotKey设置小键盘数字快捷键

Qt第三方库QHotKey设置小键盘数字快捷键

一、看了一圈没有找到可以设置小键盘的情况。 这两天在研究快捷键的使用。发现qt的里的快捷键不是全局的。找了两个第三方快捷键QHotKey，还有一个QxtGlobalShortcut。但是这两个都不能设置小键盘的数字。 比如QKeySequenceEdit （Ctrl1） 这个…
阅读更多...
springboot小型超市商品展销系统-计算机毕业设计源码01635

springboot小型超市商品展销系统-计算机毕业设计源码01635

摘 要 科技进步的飞速发展引起人们日常生活的巨大变化，电子信息技术的飞速发展使得电子信息技术的各个领域的应用水平得到普及和应用。信息时代的到来已成为不可阻挡的时尚潮流，人类发展的历史正进入一个新时代。在现实运用中，应用软件的工作…
阅读更多...
onnx基本概念

onnx基本概念

onnx基本概念 参考 文章目录 onnx基本概念Input, Output, Node, Initializer, AttributesSerialization with protobuf元数据List of available operators and domains支持的类型Opset版本Subgraphs, tests and loopsExtensibilityFunctionsShape (and Type) Inferencetools O…
阅读更多...
Fiddler抓包工具介绍

Fiddler抓包工具介绍

下载 下载:Web Debugging Proxy and Troubleshooting Tools|Fiddler 进去要填一个表 汉化版 百度网盘 请输入提取码 提取码：xq9t 下载过附件之后分别把两个文件 点开fiddler就ok了 配置https fiddler要想抓到https包(解密的),点击tools->options勾选三个对…
阅读更多...
数据结构之“双向链表”

数据结构之“双向链表”

前言 前面我们介绍了单向链表，我们这里的双向链表是为了弥补单向链表只能从头节点开始单向遍历，插入和删除节点时需要更多的操作，因为无法直接访问前一个节点。 目录 前言 一、双向链表的结构 二、实现双向链表 2.1符号定义 2.2节点创…
阅读更多...
半监督学习

半监督学习

提示：文章写完后，目录可以自动生成，如何生成可参考右边的帮助文档 目录 介绍一、Self Training自训练1、介绍2、代码示例3、参数解释 二、Label Propagation（标签传播）1、介绍2、代码示例3、参数解释 三、Label Spread…
阅读更多...
物联网工程的未来发展趋势及影响

物联网工程的未来发展趋势及影响

物联网工程是在互联网基础上的一种新兴技术，其核心思想是通过网络连接不同物体，实现智能化的交流与互动。在未来，物联网工程将继续向更多领域发展，如智能家居、智能城市、智能交通等。首先，物联网工程在智能家居领域的…
阅读更多...
最新文章

Copyright @ 2022~2023

友情链接： 我的编程经验分享 一条长河网 尧图网站开发 尧图建网站