Flink电商实时数仓(六)
交易域支付成功事务事实表
- 从topic_db业务数据中筛选支付成功的数据
- 从dwd_trade_order_detail主题中读取订单事实数据、LookUp字典表
- 关联三张表形成支付成功宽表
- 写入 Kafka 支付成功主题
执行步骤
- 设置ttl,通过Interval join实现左右流的状态管理
- 获取下单明细数据:用户必然要先下单才有可能支付成功,因此支付成功明细数据集必然是订单明细数据集的子集。要注意:Interval Join要求表中均为Append数据,即“只能新增,不能修改”,订单明细表数据生成过程中用到了left join,生成了回撤流,看似不满足Interval Join的条件。但是,回撤数据进入Kafka会以null值形式存在,如果用Kafka Connector将订单明细封装为动态表,null值会被过滤,最终得到的是相同主键存在重复数据的Append流(动态表本质上就是流),满足Interval Join的条件。
- Interval join只支持事件时间,因此数据必须携带水位线;建表时水位线的相关语法为
water for order_time as order_time - interval '5' second,这里要求数据是timestamp(3) - 原有的时间数据类型是bigint类型的ts,使用
row_time as TO_TIMESTAMP_LTZ(ts,3)这个函数即可将原有的时间数据转换为水位线所需的数据类型
- Interval join只支持事件时间,因此数据必须携带水位线;建表时水位线的相关语法为
- 筛选支付数据:
- 支付状态为支付成功
- 操作类型为update
- 构建 LookUp 字典表
- 联上述三张表形成支付成功宽表,写入 Kafka 支付成功主题
核心代码如下
public void handle(StreamExecutionEnvironment env, TableEnvironment tableEnv, String groupId) {//核心业务逻辑//1. 读取TopicDB主题数据createTopicDb(groupId,tableEnv);//2. 筛选支付成功的数据,从业务数据topic_db中filterPaymentTable(tableEnv);//3. 读取下单详情表数据, 从kafka读取数据createOrderDetailTable(tableEnv, groupId);//4. 创建base.dic字典表,从HBase维度数据中读取createBaseDic(tableEnv);//tableEnv.executeSql("select * from order_detail").print();//tableEnv.executeSql("select * from base_dic").print();//tableEnv.executeSql("select to_timestamp_ltz(ts,3) from order_detail");//5. 使用interval join 完成支付成功流和订单详情数据关联intervalJoin(tableEnv);//6. 使用lookup join完成维度退化Table resultTable = lookupJoin(tableEnv);//7. 创建upsert kafka连接器写出createKafkaSink(tableEnv);resultTable.insertInto(Constant.TOPIC_DWD_TRADE_ORDER_PAYMENT_SUCCESS).execute();}
事实表动态分流

dwd层其他的事实表都是从topic_db中去业务数据库一张表的变更数据,按照某些过滤后写入kafka的对应主题,它们处理逻辑相似且较为简单,可以结合配置表动态分流在同一个程序中处理。有点类似我们前面实现DIM层的动态配置。
- 清洗过滤和转换:判断是否满足json格式,如果满足转换为jsonObj对象
- 读取配置表数据,使用flink-cdc读取
- 转换数据格式,转换到对应bean对象中
- 配置信息广播话,然后跟主流数据进行连接
- 筛选出需要的字段
- 根据表中的sink table字段来动态写出到对应的kafka主题中
核心代码如下
public static void main(String[] args) {new DwdBaseDb().start(10019, 4, "dwd_base_db", Constant.TOPIC_DB);}@Overridepublic void handle(StreamExecutionEnvironment env, DataStreamSource<String> stream) {//核心业务逻辑//1. 读取topic_db数据//stream.print();//2. 清洗过滤和转换, jsonObjStream是主流数据SingleOutputStreamOperator<JSONObject> jsonObjStream = filterJson(stream);//jsonObjStream.print();//3. 读取配置表数据,使用flink-cdc读取,读取配置文件时并发度最好为1DataStreamSource<String> tableProcessDwd = getTableProcessDwd(env);//tableProcessDwd.print();4. 转换数据格式 string -> TableProcessDwd -> broadcastStream,广播流数据SingleOutputStreamOperator<TableProcessDwd> processDwdStream = getProcessDwdStream(tableProcessDwd);MapStateDescriptor<String, TableProcessDwd> mapStateDescriptor = new MapStateDescriptor<>("process_state", String.class, TableProcessDwd.class);BroadcastStream<TableProcessDwd> broadcastStream = processDwdStream.broadcast(mapStateDescriptor);//5. 连接主流和广播流,对主流数据进行判断是否需要保留SingleOutputStreamOperator<Tuple2<JSONObject, TableProcessDwd>> processStream = processBaseDb(jsonObjStream, broadcastStream, mapStateDescriptor);//processStream.print();//6. 筛选最后需要写出的字段SingleOutputStreamOperator<JSONObject> dataStream = filterColumns(processStream);//7. 通过sink_table的表名来动态写出到对应kafka主题//在setRecordSerializer()设置dataStream.sinkTo(FlinkSinkUtil.getKafkaSinkWithTopicName());}
gitee地址 :https://gitee.com/langpaian/gmall2023-realtime
相关文章:
Flink电商实时数仓(六)
交易域支付成功事务事实表 从topic_db业务数据中筛选支付成功的数据从dwd_trade_order_detail主题中读取订单事实数据、LookUp字典表关联三张表形成支付成功宽表写入 Kafka 支付成功主题 执行步骤 设置ttl,通过Interval join实现左右流的状态管理获取下单明细数据…...
本地部署Jellyfin影音服务器并实现远程访问内网影音库
文章目录 1. 前言2. Jellyfin服务网站搭建2.1. Jellyfin下载和安装2.2. Jellyfin网页测试 3.本地网页发布3.1 cpolar的安装和注册3.2 Cpolar云端设置3.3 Cpolar本地设置 4.公网访问测试5. 结语 1. 前言 随着移动智能设备的普及,各种各样的使用需求也被开发出来&…...
【React Native】第一个Android应用
第一个Android应用 环境TIP开发工具环境及版本要求建议官方建议 安装 Android Studio首次安装模板选择安装 Android SDK配置 ANDROID_HOME 环境变量把一些工具目录添加到环境变量 Path[可选参数] 指定版本或项目模板 运行使用 Android 模拟器编译并运行 React Native 应用修改项…...
解决IOS transform rotate后文字无法显示,backface-visibility导致@click事件失效
问题一:IOS transform rotate后文字无法显示 网上搜到可以用backface-visibility:hidden来解决,这样做文字是出来了,但是click事件无效了。 问题二:backface-visibility导致click事件失效 在Vue中使用backface-visibility和cli…...
Nature | 大型语言模型(LLM)能够产生和发现新知识吗?
大型语言模型(LLM)是基于大量数据进行预训练的超大型深度学习模型。底层转换器是一组神经网络,这些神经网络由具有自注意力功能的编码器和解码器组成。编码器和解码器从一系列文本中提取含义,并理解其中的单词和短语之间的关系。通…...
多维时序 | MATLAB实CNN-Mutilhead-Attention卷积神经网络融合多头注意力机制多变量时间序列预测
多维时序 | MATLAB实CNN-Mutilhead-Attention卷积神经网络融合多头注意力机制多变量时间序列预测 目录 多维时序 | MATLAB实CNN-Mutilhead-Attention卷积神经网络融合多头注意力机制多变量时间序列预测预测效果基本介绍模型描述程序设计参考资料 预测效果 基本介绍 多维时序 | …...
Nature 新研究发布,GPT 驱动的机器人化学家能够自行设计和进行实验,这对科研意味着什么?
文章目录 前言揭秘Coscientist不到四分钟,设计并改进了程序能力越大,责任越大 前言 有消息称,AI 大模型 “化学家” 登 Nature 能够自制阿司匹林、对乙酰氨基酚、布洛芬,甚至连复杂的钯催化交叉偶联反应,也能完成。 …...
Ai画板原理
在创建时画板可以选择数量和排列方式 也可以采用这个图片左上的画板工具,选择画板在其他地方画框即可生成,同时可以在属性框中可以修改尺寸大小 选择全部重新排列可以进行创建时的布局...
【hacker送书第11期】Python数据分析从入门到精通
探索数据世界,揭示未来趋势 《Python数据分析从入门到精通》是你掌握Python数据分析的理想选择。本书深入讲解核心工具如pandas、matplotlib和numpy,助您轻松处理和理解复杂数据。 通过matplotlib、seaborn和创新的pyecharts,本书呈现生动直…...
华为OD机试 - 精准核酸检测(Java JS Python C)
在线OJ刷题 题目详情 - 精准核酸检测 - Hydro 题目描述 为了达到新冠疫情精准防控的需要,为了避免全员核酸检测带来的浪费,需要精准圈定可能被感染的人群。 现在根据传染病流调以及大数据分析,得到了每个人之间在时间、空间上是否存在轨迹交叉。 现在给定一组确诊人员编…...
智能优化算法应用:基于材料生成算法3D无线传感器网络(WSN)覆盖优化 - 附代码
智能优化算法应用:基于材料生成算法3D无线传感器网络(WSN)覆盖优化 - 附代码 文章目录 智能优化算法应用:基于材料生成算法3D无线传感器网络(WSN)覆盖优化 - 附代码1.无线传感网络节点模型2.覆盖数学模型及分析3.材料生成算法4.实验参数设定5.算法结果6.…...
【MySQL】:超详细MySQL完整安装和配置教程
🎥 屿小夏 : 个人主页 🔥个人专栏 : MySQL从入门到进阶 🌄 莫道桑榆晚,为霞尚满天! 文章目录 📑前言一. MySQL数据库1.1 版本1.2 下载1.3 安装1.4 客户端连接 🌤️全篇总…...
OpenAI亲授ChatGPT “屠龙术”!官方Prompt 工程指南来啦
应该如何形容 Prompt 工程呢?对于一个最开始使用 ChatGPT 的新人小白,面对据说参数量千亿万亿的庞然巨兽,Prompt 神秘的似乎像某种献祭:我扔进去几句话,等待聊天窗口后的“智慧生命”给我以神谕。 然而,上…...
最新ChatGPT商业运营网站程序源码,支持Midjourney绘画,GPT语音对话+DALL-E3文生图+文档对话总结
一、前言 SparkAi创作系统是基于ChatGPT进行开发的Ai智能问答系统和Midjourney绘画系统,支持OpenAI-GPT全模型国内AI全模型。本期针对源码系统整体测试下来非常完美,可以说SparkAi是目前国内一款的ChatGPT对接OpenAI软件系统。那么如何搭建部署AI创作Ch…...
经验 | IDEA常用快捷键
1、编辑(Editing) Ctrl Space 基本的代码完成(类、方法、属性) Ctrl Alt Space 快速导入任意类 Ctrl Shift Enter 语句完成 Ctrl P 参数信息(在方法中调用参数) Ctrl Q 快速查看文档 Shift F…...
spark中 write.csv时, 添加第一行的标题title
在 Spark 中使用 write.csv 写入 CSV 文件时,默认情况下是不会在文件中添加标题行的。但是,你可以通过设置 header 选项来控制是否包含标题行。 下面是一个示例: val data Seq((1, "John", 28),(2, "Alice", 22),(3, …...
HTML美化网页
使用CSS3美化的原因 用css美化页面文本,使页面漂亮、美观、吸引用户 可以更好的突出页面的主题内容,使用户第一眼可以看到页面主要内容 具有良好的用户体验 <span>标签 作用 能让某几个文字或者某个词语凸显出来 有效的传递页面信息用css美化页面文本,使页面漂…...
nn.LSTM个人记录
简介 nn.LSTM参数 torch.nn.lstm(input_size, "输入的嵌入向量维度,例如每个单词用50维向量表示,input_size就是50"hidden_size, "隐藏层节点数量,也是输出的嵌入向量维度"num_layers, "lstm 隐层的层数,默认…...
vr虚拟高压电器三维仿真展示更立体全面
VR工业虚拟仿真软件的应用价值主要体现在以下几个方面: 降低成本:通过VR技术进行产品设计和开发,可以在虚拟环境中进行,从而减少对物理样机的依赖,降低试错成本和时间。此外,利用VR技术构建的模拟场景使用方…...
轮廓平滑方法
目录 1. 形态学操作 2. 边缘平滑化 3. 轮廓近似 python 有回归线平滑 2D 轮廓 1. 形态学操作 利用形态学操作(例如腐蚀、膨胀、开运算、闭运算等)可以使分割边界更加平滑和连续。腐蚀可以消除小的不连续区域,膨胀可以填充空洞࿰…...
浏览器访问 AWS ECS 上部署的 Docker 容器(监听 80 端口)
✅ 一、ECS 服务配置 Dockerfile 确保监听 80 端口 EXPOSE 80 CMD ["nginx", "-g", "daemon off;"]或 EXPOSE 80 CMD ["python3", "-m", "http.server", "80"]任务定义(Task Definition&…...
Spark 之 入门讲解详细版(1)
1、简介 1.1 Spark简介 Spark是加州大学伯克利分校AMP实验室(Algorithms, Machines, and People Lab)开发通用内存并行计算框架。Spark在2013年6月进入Apache成为孵化项目,8个月后成为Apache顶级项目,速度之快足见过人之处&…...
React hook之useRef
React useRef 详解 useRef 是 React 提供的一个 Hook,用于在函数组件中创建可变的引用对象。它在 React 开发中有多种重要用途,下面我将全面详细地介绍它的特性和用法。 基本概念 1. 创建 ref const refContainer useRef(initialValue);initialValu…...
UE5 学习系列(三)创建和移动物体
这篇博客是该系列的第三篇,是在之前两篇博客的基础上展开,主要介绍如何在操作界面中创建和拖动物体,这篇博客跟随的视频链接如下: B 站视频:s03-创建和移动物体 如果你不打算开之前的博客并且对UE5 比较熟的话按照以…...
华为云Flexus+DeepSeek征文|DeepSeek-V3/R1 商用服务开通全流程与本地部署搭建
华为云FlexusDeepSeek征文|DeepSeek-V3/R1 商用服务开通全流程与本地部署搭建 前言 如今大模型其性能出色,华为云 ModelArts Studio_MaaS大模型即服务平台华为云内置了大模型,能助力我们轻松驾驭 DeepSeek-V3/R1,本文中将分享如何…...
如何在最短时间内提升打ctf(web)的水平?
刚刚刷完2遍 bugku 的 web 题,前来答题。 每个人对刷题理解是不同,有的人是看了writeup就等于刷了,有的人是收藏了writeup就等于刷了,有的人是跟着writeup做了一遍就等于刷了,还有的人是独立思考做了一遍就等于刷了。…...
Go 语言并发编程基础:无缓冲与有缓冲通道
在上一章节中,我们了解了 Channel 的基本用法。本章将重点分析 Go 中通道的两种类型 —— 无缓冲通道与有缓冲通道,它们在并发编程中各具特点和应用场景。 一、通道的基本分类 类型定义形式特点无缓冲通道make(chan T)发送和接收都必须准备好࿰…...
C++:多态机制详解
目录 一. 多态的概念 1.静态多态(编译时多态) 二.动态多态的定义及实现 1.多态的构成条件 2.虚函数 3.虚函数的重写/覆盖 4.虚函数重写的一些其他问题 1).协变 2).析构函数的重写 5.override 和 final关键字 1&#…...
JavaScript基础-API 和 Web API
在学习JavaScript的过程中,理解API(应用程序接口)和Web API的概念及其应用是非常重要的。这些工具极大地扩展了JavaScript的功能,使得开发者能够创建出功能丰富、交互性强的Web应用程序。本文将深入探讨JavaScript中的API与Web AP…...
关于uniapp展示PDF的解决方案
在 UniApp 的 H5 环境中使用 pdf-vue3 组件可以实现完整的 PDF 预览功能。以下是详细实现步骤和注意事项: 一、安装依赖 安装 pdf-vue3 和 PDF.js 核心库: npm install pdf-vue3 pdfjs-dist二、基本使用示例 <template><view class"con…...
