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. 形态学操作 利用形态学操作(例如腐蚀、膨胀、开运算、闭运算等)可以使分割边界更加平滑和连续。腐蚀可以消除小的不连续区域,膨胀可以填充空洞࿰…...
vscode(仍待补充)
写于2025 6.9 主包将加入vscode这个更权威的圈子 vscode的基本使用 侧边栏 vscode还能连接ssh? debug时使用的launch文件 1.task.json {"tasks": [{"type": "cppbuild","label": "C/C: gcc.exe 生成活动文件"…...
理解 MCP 工作流:使用 Ollama 和 LangChain 构建本地 MCP 客户端
🌟 什么是 MCP? 模型控制协议 (MCP) 是一种创新的协议,旨在无缝连接 AI 模型与应用程序。 MCP 是一个开源协议,它标准化了我们的 LLM 应用程序连接所需工具和数据源并与之协作的方式。 可以把它想象成你的 AI 模型 和想要使用它…...
ffmpeg(四):滤镜命令
FFmpeg 的滤镜命令是用于音视频处理中的强大工具,可以完成剪裁、缩放、加水印、调色、合成、旋转、模糊、叠加字幕等复杂的操作。其核心语法格式一般如下: ffmpeg -i input.mp4 -vf "滤镜参数" output.mp4或者带音频滤镜: ffmpeg…...
ElasticSearch搜索引擎之倒排索引及其底层算法
文章目录 一、搜索引擎1、什么是搜索引擎?2、搜索引擎的分类3、常用的搜索引擎4、搜索引擎的特点二、倒排索引1、简介2、为什么倒排索引不用B+树1.创建时间长,文件大。2.其次,树深,IO次数可怕。3.索引可能会失效。4.精准度差。三. 倒排索引四、算法1、Term Index的算法2、 …...
【python异步多线程】异步多线程爬虫代码示例
claude生成的python多线程、异步代码示例,模拟20个网页的爬取,每个网页假设要0.5-2秒完成。 代码 Python多线程爬虫教程 核心概念 多线程:允许程序同时执行多个任务,提高IO密集型任务(如网络请求)的效率…...
【学习笔记】深入理解Java虚拟机学习笔记——第4章 虚拟机性能监控,故障处理工具
第2章 虚拟机性能监控,故障处理工具 4.1 概述 略 4.2 基础故障处理工具 4.2.1 jps:虚拟机进程状况工具 命令:jps [options] [hostid] 功能:本地虚拟机进程显示进程ID(与ps相同),可同时显示主类&#x…...
云原生玩法三问:构建自定义开发环境
云原生玩法三问:构建自定义开发环境 引言 临时运维一个古董项目,无文档,无环境,无交接人,俗称三无。 运行设备的环境老,本地环境版本高,ssh不过去。正好最近对 腾讯出品的云原生 cnb 感兴趣&…...
CSS设置元素的宽度根据其内容自动调整
width: fit-content 是 CSS 中的一个属性值,用于设置元素的宽度根据其内容自动调整,确保宽度刚好容纳内容而不会超出。 效果对比 默认情况(width: auto): 块级元素(如 <div>)会占满父容器…...
Java求职者面试指南:计算机基础与源码原理深度解析
Java求职者面试指南:计算机基础与源码原理深度解析 第一轮提问:基础概念问题 1. 请解释什么是进程和线程的区别? 面试官:进程是程序的一次执行过程,是系统进行资源分配和调度的基本单位;而线程是进程中的…...
云原生安全实战:API网关Kong的鉴权与限流详解
🔥「炎码工坊」技术弹药已装填! 点击关注 → 解锁工业级干货【工具实测|项目避坑|源码燃烧指南】 一、基础概念 1. API网关(API Gateway) API网关是微服务架构中的核心组件,负责统一管理所有API的流量入口。它像一座…...
