【实时数据仓库建模】Kafka+Flink实战 【实时数据仓库建模】KafkaFlink实战一、引言二、核心步骤2.1 需求分析与建模2.2 数据表创建2.3 Kafka生产者 - Nodered模拟订单数据2.4 Flink处理 - 从ODS到DWD再到DWS三、源码3.1 Nodered模拟订单发送到Kafka源码一、引言1、核心通过KafkaFlink构建实时数据仓库解决传统数据仓库“慢”的问题将数据从产生到分析的延迟缩短到秒级或分钟级Kafka实时数据管道负责收集、存储和传输海量流数据具备高吞吐量、低延迟、可持久化等特性Flink实时计算引擎负责对Kafka中的流数据进行清洗、转换、聚合支持精确一次Exactly-Once语义确保数据一致性。2、最终效果展示本文参考链接https://blog.csdn.net/2501_91888447/article/details/155677083本文与参考链接的区别是将参考链接中每一步进行实现实战并截图说明其中python构造数据部分换成Nodered实现本文将以电商实时销售额监控为例构建一个端到端的实时数据仓库。最终实现实时 dashboard展示每小时销售额、订单量、TOP 10商品更新频率1分钟数据链路订单数据从产生Kafka生产者→ 存储Kafka topic→ 处理Flink→ 分析ClickHouse→ 展示Grafana全程延迟≤1分钟。3、环境与工具清单4、 环境搭建步骤简要安装Kafka与ZooKeeper参考Kafka官方文档安装ClickHouse参考ClickHouse官方文档安装Grafana参考Grafana官方文档安装MySQL参考MySQL官方文档安装Flink参考Flink官方文档注意这里的Flink需要开启SQL Gateway服务并支持Kafka和ClickHouse插件参考实战链接【Flink on Kubernetes部署详细教程】安装Nodered参考Nodered官方文档回到目录二、核心步骤2.1 需求分析与建模需求分析本次实战的需求是电商平台实时销售额监控具体如下实时统计每小时的总销售额、总订单量TOP商品每小时销售额前10的商品数据延迟从订单产生到 dashboard 展示≤1分钟数据准确性确保数据不丢失、不重复精确一次语义。数据建模实时数据仓库分层设计强调低延迟和流处理特性。本文采用经典的“三层模型”层级全称作用说明存储介质ODS操作数据存储保留原始数据如订单、用户行为不做任何清洗方便回溯Kafka TopicDWD数据仓库明细层对ODS层数据进行清洗过滤无效数据、补全关联维度表生成干净的明细数据Kafka TopicDWS数据仓库汇总层按主题如销售额、“订单量”进行实时聚合生成汇总数据支持快速查询ClickHouse 表ODS层原始数据存储ODS层的目标是保存原始数据避免数据丢失。本次实战中ODS层对应Kafka的order_topic存储从生产者发送的原始订单数据。DWD层明细数据清洗DWD层的目标是干净、完整对应Kafka的dwd_order_topic存储清洗后的明细数据。主要做两件事过滤无效数据比如取消的订单status‘canceled’不需要统计关联维度表补全商品名称product_name方便后续分析。DWS层汇总数据计算DWS层的目标是快速查询按时间窗口如1小时对DWD层数据进行聚合生成汇总结果。本次实战中DWS层包含两张表dws_hourly_sales每小时总销售额、总订单量dws_hourly_top_products每小时销售额前10的商品。这些汇总数据存储在ClickHouse中因为ClickHouse是列式存储数据库支持秒级查询非常适合实时分析。回到目录2.2 数据表创建创建MySQL维度表CREATE DATABASE db_test;CREATE TABLE db_test.product (product_id INT PRIMARY KEY AUTO_INCREMENT,product_name VARCHAR(100) NOT NULL,price DECIMAL(10,2) NOT NULL);– 插入测试数据INSERT INTO db_test.product (product_name, price) VALUES(‘iPhone 15’, 7999.00),(‘MacBook Pro’, 14999.00),(‘iPad Pro’, 6999.00),(‘Apple Watch Series 9’, 2999.00),(‘AirPods Pro 2’, 1999.00);回到目录2.3 Kafka生产者 - Nodered模拟订单数据首先我们需要一个Kafka生产者模拟电商平台的订单生成。这里用Nodered实现此时能看到Kafka的order_topic中已经有了模拟的订单数据点击跳转查看源码回到目录2.4 Flink处理 - 从ODS到DWD再到DWSFlink是本次实战的“核心引擎”负责处理从Kafka读取的数据完成DWD层清洗和DWS层聚合。我们使用Flink SQL更易上手适合批量处理来实现。在DBEaver连接Flink SQL客户端或直接在Flink SQL客户端执行sql中创建ODS层表ods_order关联Kafka的order_topicCREATE TABLE IF NOT EXISTS ods_order(order_id INT, user_id INT, product_id INT, amount DECIMAL(10,2), create_time STRING, status STRING, proc_time AS PROCTIME(), event_time AS TO_TIMESTAMP(create_time,yyyy-MM-ddTHH:mm:ss.SSSZ), WATERMARK FOR event_time AS event_time - INTERVAL5SECOND)WITH(connectorkafka,topicorder_topic,properties.bootstrap.servers10.30.2.95:9092,properties.group.idflink_ods_order_group,formatjson,scan.startup.modelatest-offset);order_id订单ID user_id 用户ID product_id商品ID amount 订单金额 create_time 订单创建时间status 订单状态event_time事件时间即订单的实际创建时间从create_time字段转换而来WATERMARK水印用于处理事件时间的延迟比如订单数据因网络问题延迟5秒到达水印会等待5秒再关闭窗口确保数据不丢失。MySQL商品信息维度表与MySQL结构一致CREATE TABLE IF NOT EXISTS product_dim(product_id INT PRIMARY KEY NOT ENFORCED, product_name STRING, price DECIMAL(10,2))WITH(connectorjdbc,urljdbc:mysql://10.30.2.85:3306/db_test,table-nameproduct,usernamejxdev,passwordahjuxin_2026,lookup.cache.max-rows1000,lookup.cache.ttl1h);DWD层表dwd_order清洗后的明细数据并写入kafkaCREATE TABLE IF NOT EXISTS dwd_order(order_id INT, user_id INT, product_id INT, product_name STRING, amount DECIMAL(10,2), create_time STRING, status STRING, event_time TIMESTAMP(3), -- 关键定义事件时间与水印把event_time转为时间属性 WATERMARK FOR event_time AS event_time - INTERVAL5SECOND)WITH(connectorkafka,topicdwd_order_topic,properties.bootstrap.servers10.30.2.95:9092,properties.group.idflink_dws_hourly_sales_group,formatjson,sink.partitionerround-robin);-- 插入数据到DWD层过滤取消订单关联商品信息维度表 INSERT INTO dwd_order SELECT o.order_id,o.user_id,o.product_id,p.product_name, o.amount,o.create_time,o.status, o.event_time FROM ods_order o JOIN product_dim FOR SYSTEM_TIME AS OF o.proc_time p ON o.product_idp.product_id WHERE o.statuscanceled;FOR SYSTEM_TIME AS OF o.proc_timeFlink的时态表关联Temporal Table Join用于关联维度表的快照避免维度数据更新导致的不一致dwd_order_topicDWD层数据存储的Kafka Topic后续可以用于其他分析如用户行为分析。创建DWS层表实时聚合计算DWS层是实时数据仓库的“出口”负责生成汇总数据供可视化工具查询。本次实战中创建两张DWS层表① 每小时销售额与订单量dws_hourly_sales② 每小时TOP 10商品dws_hourly_top_products--5、创建DWS层表存储每小时销售额汇总数据并写入ClickHouse表 CREATE TABLE IF NOT EXISTS dws_hourly_sales(hourSTRING, -- 小时格式yyyy-MM-dd HH:00 total_sales DECIMAL(10,2), -- 总销售额 total_orders INT, -- 总订单量 window_start TIMESTAMP(3), -- 窗口开始时间 window_end TIMESTAMP(3)-- 窗口结束时间)WITH(connectorclickhouse,urljdbc:ch://10.30.2.98:8123,database-namebench,table-namedws_hourly_sales,usernamedefault,passwordHuayu_2025,sink.batch-size1000,sink.flush-interval1000);-- 插入数据到DWS层每小时汇总销售额与订单量 INSERT INTO dws_hourly_sales SELECT DATE_FORMAT(window_start,yyyy-MM-dd HH:00)AShour, SUM(amount)AS total_sales, CAST(COUNT(DISTINCT order_id)AS INT)AS total_orders, window_start, window_end FROM TABLE(TUMBLE(TABLE dwd_order, DESCRIPTOR(event_time), INTERVAL1HOUR))GROUP BY window_start, window_end;--6、创建DWS层表存储每小时TOP10商品数据写入ClickHouse CREATE TABLE dws_hourly_top_products(hourSTRING, -- 小时格式yyyy-MM-dd HH:00 product_id INT, -- 商品ID product_name STRING, -- 商品名称 sales DECIMAL(10,2), -- 商品销售额rankINT, -- 排名1-10 window_start TIMESTAMP(3), -- 窗口开始时间 window_end TIMESTAMP(3), -- 窗口结束时间 PRIMARY KEY(hour, product_id)NOT ENFORCED)WITH(connectorclickhouse,urljdbc:ch://10.30.2.98:8123,database-namebench,table-namedws_hourly_top_products,usernamedefault,passwordHuayu_2025,sink.batch-size1000,sink.flush-interval1000);-- 插入数据到DWS层每小时TOP10商品 INSERT INTO dws_hourly_top_products SELECT DATE_FORMAT(window_start,yyyy-MM-dd HH:00)AShour, product_id, product_name, sales, CAST(rn AS INT)asrank, window_start, window_end FROM(SELECT product_id, product_name, SUM(amount)AS sales, window_start, window_end, ROW_NUMBER()OVER(PARTITION BY window_start ORDER BY SUM(amount)DESC)AS rn FROM TABLE(TUMBLE(TABLE dwd_order, DESCRIPTOR(event_time), INTERVAL1HOUR))GROUP BY product_id, product_name, window_start, window_end)WHERE rn10;说明滚动窗口Tumbling Window每1小时一个窗口窗口之间不重叠如14:00-15:00、15:00-16:00适合固定时间间隔的汇总RANK()函数用于计算每个商品在窗口内的排名降序PARTITION BY window_start表示按窗口分组ORDER BY SUM(amount) DESC表示按销售额降序排列ClickHouse存储ClickHouse的列式存储和向量查询引擎使得汇总数据的查询速度非常快秒级适合实时展示。验证表是否创建成功show tables;执行成功后可以看到flink客户端有正在运行中的任务回到目录三、源码3.1 Nodered模拟订单发送到Kafka源码回到目录[{id:67428b8f6878b23b,type:inject,z:067b766577a0b317,name:,props:[{p:payload},{p:topic,vt:str}],repeat:,crontab:,once:false,onceDelay:0.1,topic:,payload:,payloadType:date,x:90,y:60,wires:[[51cdba257295fe80]]},{id:51cdba257295fe80,type:function,z:067b766577a0b317,name:初始化,func:// 商品列表与MySQL维度表一致\nmsg.products [\n{product_id: 1, product_name: iPhone 15, price: 7999.00},\n{product_id: 2, product_name: MacBook Pro, price: 14999.00},\n{product_id: 3, product_name: iPad Pro, price: 6999.00},\n{product_id: 4, product_name: Apple Watch Series 9, price: 2999.00},\n{product_id: 5, product_name: AirPods Pro 2, price: 1999.00}\n]\nmsg.orderStatusList [completed, pending, canceled];//订单状态\nreturn msg;,outputs:1,timeout:0,noerr:0,initialize:,finalize:,libs:[],x:230,y:60,wires:[[fc8302a4d3979c14]]},{id:8582719c984a6e2b,type:function,z:067b766577a0b317,name:生成模拟订单,func:// 生成模拟订单\nconst product msg.products[randomInt(0,4)];//订单选购商品\nmsg.payload {\norder_id: randomInt(100000, 999999), // 随机订单ID\nuser_id: randomInt(1, 10000), // 随机用户ID\nproduct_id: product.product_id,// 商品ID来自商品列表\namount: Number((randomInt(1,5)*product.price).toFixed(2)), // 订单金额数量×单价\ncreate_time: new Date(), // 订单创建时间当前时间\nstatus: msg.orderStatusList[randomInt(0, 2)] // 随机订单状态\n}\nreturn msg;\n/** 生成区间[min,max]内的随机数int */\nfunction randomInt(min, max) {\nconst ran Math.floor(Math.random() * (max - min 1)) min;\nif (ran max){ return max; }\nif (ran min) { return min; }\nreturn ran;\n},outputs:1,timeout:0,noerr:0,initialize:,finalize:,libs:[],x:380,y:140,wires:[[52d6654949bb37c1,8fc61fbcdcc14c6b]]},{id:8fc61fbcdcc14c6b,type:delay,z:067b766577a0b317,name:随机延迟1-5秒,pauseType:random,timeout:5,timeoutUnits:seconds,rate:1,nbRateUnits:1,rateUnits:second,randomFirst:1,randomLast:5,randomUnits:seconds,drop:false,allowrate:false,outputs:1,x:620,y:140,wires:[[fc8302a4d3979c14]]},{id:fc8302a4d3979c14,type:counter-loop,z:067b766577a0b317,name:,counter:cid,counterType:msg,reset:false,resetValue:value-null,initial:0,initialType:num,operator:lt,termination:100,terminationType:num,increment:1,incrementType:num,x:440,y:60,wires:[[8ed5ca6d35c83157],[8582719c984a6e2b]]},{id:52d6654949bb37c1,type:rdkafka out,z:067b766577a0b317,name:,topic:order_topic,key:,partition:-1,broker:c331c6a48d89a75a,x:610,y:180,wires:[]},{id:8ed5ca6d35c83157,type:function,z:067b766577a0b317,name:结束,func:node.warn(\模拟订单发送结束共产生\msg.cid\条订单\);\nreturn msg;,outputs:1,timeout:0,noerr:0,initialize:,finalize:,libs:[],x:590,y:40,wires:[[]]},{id:c331c6a48d89a75a,type:kafka-broker,broker:10.30.2.95:9092,clientid:}]回到目录