大数据-玩转数据-Flink SQL编程实战 (热门商品TOP N)
一、需求描述
每隔30min 统计最近 1hour的热门商品 top3, 并把统计的结果写入到mysql中。
二、需求分析
- 1.统计每个商品的点击量, 开窗
- 2.分组窗口分组
- 3.over窗口
三、需求实现
3.1、创建数据源示例
input/UserBehavior.csv
543462,1715,1464116,pv,1511658000
662867,2244074,1575622,pv,1511658000
561558,3611281,965809,pv,1511658000
894923,3076029,1879194,pv,1511658000
834377,4541270,3738615,pv,1511658000
315321,942195,4339722,pv,1511658000
625915,1162383,570735,pv,1511658000
578814,176722,982926,pv,1511658000
873335,1256540,1451783,pv,1511658000
429984,4625350,2355072,pv,1511658000
866796,534083,4203730,pv,1511658000
937166,321683,2355072,pv,1511658000
156905,2901727,3001296,pv,1511658000
758810,5109495,1575622,pv,1511658000
107304,111477,4173315,pv,1511658000
452437,3255022,5099474,pv,1511658000
813974,1332724,2520771,buy,1511658000
524395,3887779,2366905,pv,1511658000
3.2、创建目标表
CREATE DATABASE flink_sql; //创建flink_sql库
USE flink_sql;
DROP TABLE IF EXISTS `hot_item`;
CREATE TABLE `hot_item` (`w_end` timestamp NOT NULL,`item_id` bigint(20) NOT NULL,`item_count` bigint(20) NOT NULL,`rk` bigint(20) NOT NULL,PRIMARY KEY (`w_end`,`rk`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8;
3.3、导入JDBC Connector依赖
<!-- 导入JDBC Connector依赖 --><dependency><groupId>org.apache.flink</groupId><artifactId>flink-connector-jdbc_${scala.binary.version}</artifactId><version>${flink.version}</version></dependency>
3.4、代码实现
package com.atguigu.flink.java.chapter_12;import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;/*** @Author lizhenchao@atguigu.cn* @Date 2021/1/31 9:11*/
public class Flink01_HotItem_TopN {public static void main(String[] args) {StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();env.setParallelism(2);StreamTableEnvironment tenv = StreamTableEnvironment.create(env);// 使用sql从文件读取数据tenv.executeSql("create table user_behavior(" +" user_id bigint, " +" item_id bigint, " +" category_id int, " +" behavior string, " +" ts bigint, " +" event_time as to_timestamp(from_unixtime(ts, 'yyyy-MM-dd HH:mm:ss')), " +" watermark for event_time as event_time - interval '5' second " +")with(" +" 'connector'='filesystem', " +" 'path'='input/UserBehavior.csv', " +" 'format'='csv')");// 每隔 10m 统计一次最近 1h 的热门商品 top// 1. 计算每每个窗口内每个商品的点击量Table t1 = tenv.sqlQuery("select " +" item_id, " +" hop_end(event_time, interval '10' minute, interval '1' hour) w_end," +" count(*) item_count " +"from user_behavior " +"where behavior='pv' " +"group by hop(event_time, interval '10' minute, interval '1' hour), item_id");tenv.createTemporaryView("t1", t1);// 2. 按照窗口开窗, 对商品点击量进行排名Table t2 = tenv.sqlQuery("select " +" *," +" row_number() over(partition by w_end order by item_count desc) rk " +"from t1");tenv.createTemporaryView("t2", t2);// 3. 取 top3Table t3 = tenv.sqlQuery("select " +" item_id, w_end, item_count, rk " +"from t2 " +"where rk<=3");// 4. 数据写入到mysql// 4.1 创建输出表tenv.executeSql("create table hot_item(" +" item_id bigint, " +" w_end timestamp(3), " +" item_count bigint, " +" rk bigint, " +" PRIMARY KEY (w_end, rk) NOT ENFORCED)" +"with(" +" 'connector' = 'jdbc', " +" 'url' = 'jdbc:mysql://hadoop162:3306/flink_sql?useSSL=false', " +" 'table-name' = 'hot_item', " +" 'username' = 'root', " +" 'password' = 'aaaaaa' " +")");// 4.2 写入到输出表t3.executeInsert("hot_item");}
}
执行结果:
四、总结
Flink 使用 OVER 窗口条件和过滤条件相结合以进行 Top-N 查询。利用 OVER 窗口的 PARTITION BY 子句的功能,Flink 还支持逐组 Top-N 。 例如,每个类别中实时销量最高的前五种产品。批处理表和流处理表都支持基于SQL的 Top-N 查询。
流处理模式需注意: TopN 查询的结果会带有更新。 Flink SQL 会根据排序键对输入的流进行排序;若 top N 的记录发生了变化,变化的部分会以撤销、更新记录的形式发送到下游。 推荐使用一个支持更新的存储作为 Top-N 查询的 sink 。另外,若 top N 记录需要存储到外部存储,则结果表需要拥有与 Top-N 查询相同的唯一键。
相关文章:

大数据-玩转数据-Flink SQL编程实战 (热门商品TOP N)
一、需求描述 每隔30min 统计最近 1hour的热门商品 top3, 并把统计的结果写入到mysql中。 二、需求分析 1.统计每个商品的点击量, 开窗2.分组窗口分组3.over窗口 三、需求实现 3.1、创建数据源示例 input/UserBehavior.csv 543462,1715,1464116,pv,1511658000 662867,22…...

python中实现定时任务的几种方案
目录 while True: sleep()Timeloop库threading.Timersched模块schedule模块APScheduler框架Celery框架数据流工具Apache Airflow概述Airflow 核心概念Airflow 的架构 总结以下几种方案实现定时任务,可根据不同需求去使用不同方案。 while True: sleep() 利用whil…...

AcWing算法提高课-5.6.1同余方程
宣传一下 算法提高课整理 CSDN个人主页:更好的阅读体验 原题链接 题目描述 求关于 x x x 的同余方程 a x ≡ 1 ( m o d b ) ax ≡ 1 \pmod b ax≡1(modb) 的最小正整数解。 输入格式 输入只有一行,包含两个正整数 a , b a,b a,b,用一…...

Docker Tutorial
什么是Docker 为每个应用提供完全隔离的运行环境 Dockerfile, Image,Container Image: 相当于虚拟机的快照(snapshot)里面包含了我们需要部署的应用程序以及替它所关联的所有库。通过image,我们可以创建很…...

平面图—简单应用
平面图:若一个图𝐺能画在平面𝑆上,且使𝐺的边仅在端点处相交,则称图𝐺为可嵌入平面𝑆,𝐺称为可平面图,简称为平面图。 欧拉公式:设有…...

安装JDK(Java SE Development Kit)超详细教程
文章时间 : 2023-10-04 1. 下载地址 直接去下载地址:https://www.oracle.com/java/technologies/downloads/ (需要翻墙,不想翻墙或者不想注册oracel账号的,直接去我的阿里云盘) 阿里云盘:http…...

KUKA机器人通过3点法设置工作台基坐标系的具体方法
KUKA机器人通过3点法设置工作台基坐标系的具体方法 具体方法和步骤可参考以下内容: 进入主菜单界面,依次选择“投入运行”—“测量”—基坐标,选择“3点法”, 在系统弹出的基坐标编辑界面,给基座标编号为3,命名为table1,然后单击“继续”按钮,进行下一步操作, 在弹出的…...

以太网的MAC层
以太网的MAC层 一、硬件地址 局域网中,硬件地址又称物理地址或MAC地址(因为用在MAC帧),它是局域网上每一台计算机中固化在适配器的ROM中的地址。 关于地址问题,有这样的定义:“名字指出我们所要寻…...

Hadoop启动后jps发现没有DateNode解决办法
多次使用 Hadoop namenode -format 格式化节点后DateNode丢失 找到hadoop配置文件core-site.xml查找tmp路径 进入该路径,使用rm -rf data删除data文件 再次使用Hadoop namenode -format 格式化后jps后出现DateNode节点...

VUE3照本宣科——应用实例API与setup
VUE3照本宣科——应用实例API与setup 前言一、应用实例API1.createApp()2.app.use()3.app.mount() 二、setup 前言 👨💻👨🌾📝记录学习成果,以便温故而知新 “VUE3照本宣科”是指照着中文官网和菜鸟教…...
json/js对象的key有什么区别?
1.对于JS对象来说 一个js对象如果是这样的 obj {"0": "小明","0name": "小明明", "": 18,"¥": "哈哈"," ": "爱好广泛" }对于js对象来说,有时候key是不…...

极大似然估计概念的理解——统计学习方法
目录 1.最大似然估计的概念的理解1 2.最大似然估计的概念的理解2 3.最大似然估计的概念的理解3 4.例子 1.最大似然估计的概念的理解1 最大似然估计是一种概率论在统计学上的概念,是参数估计的一种方法。给定观测数据来评估模型参数。也就是模型已知,参…...

python模拟表格任意输入位置
在表格里输入数值,要任意位置,我找到了好方法: input输入 1. 行 2. 列输入:1 excel每行输入文字input输入位置 3.2 表示输入位置在:3行个列是要实现一个类似于 Excel 表格的输入功能,并且希望能够指定输入…...

如何限制文件只能通过USB打印机打印,限制打印次数和时限并且无法在打印前查看或编辑内容
在今天这个高度信息化的时代,文档打印已经成为日常工作中不可或缺的一部分。然而,这也带来了诸多安全风险,如文档被篡改、知识产权被侵犯以及信息泄露等。为了解决这些问题,只印应运而生。作为一款独特的软件工具,只印…...

车牌文本检测与识别:License Plate Recognition Based On Multi-Angle View Model
论文作者:Dat Tran-Anh,Khanh Linh Tran,Hoai-Nam Vu 作者单位:Thuyloi University;Posts and Telecommunications Institute of Technology 论文链接:http://arxiv.org/abs/2309.12972v1 内容简介: 1)方向&#x…...
Blender中的4种视图着色模式
Blender中有四种主要的视图着色模式:线框、实体、Look Dev和渲染。它们的主要区别如下: - 线框模式只显示物体的边缘(线框),可以让您看到场景中的所有物体,也可以调整线框的颜色和背景的颜色。 - 实…...

Flutter项目安装到Android手机一直显示在assembledebug
问题 Flutter项目安装到Android手机一直显示在assembledebug 原因 网络不好,gradle依赖下载不下来 解决方案 修改如下的文件 gradle-wrapper.properties 使用腾讯提供的gradle镜像下载 distributionUrlhttps://mirrors.cloud.tencent.com/gradle/gradle-7.5…...
数据挖掘实验(二)数据预处理【等深分箱与等宽分箱】
一、分箱平滑的原理 (1)分箱方法 在分箱前,一定要先排序数据,再将它们分到等深(等宽)的箱中。 常见的有两种分箱方法:等深分箱和等宽分箱。 等深分箱:按记录数进行分箱࿰…...
Vue2 第一次学习
本章为超级浓缩版,文章过于短,方便复习使用哦~ 文章目录 1. 简单引入 vue.js2. 指令2.1 事件绑定指令 v-on (简写 )2.2 内容渲染指令2.3 双向绑定指令 v-model2.4 属性绑定指令 v-bind (简写 : )2.5 条件渲染指令2.6 循环指令 v-for 3. vue 其他知识3.1 侦听器 watch3.2 计算属…...

tiny模式基本原理整合
【Tiny模式】的基本构成 M【首头在首位】 U【/】 V【HTTP/】 Host H【真实ip】 XH \r回车 \n换行 \t制表 \ 空格 一个基本的模式构成 [method] [uri] [version]\r\nHost: [host]\r\n[method] [uri] [version]\r\nHost: [host]\r\n 检测顺序 http M H XH 有些地区 XH H M 我这边…...
在鸿蒙HarmonyOS 5中实现抖音风格的点赞功能
下面我将详细介绍如何使用HarmonyOS SDK在HarmonyOS 5中实现类似抖音的点赞功能,包括动画效果、数据同步和交互优化。 1. 基础点赞功能实现 1.1 创建数据模型 // VideoModel.ets export class VideoModel {id: string "";title: string ""…...

【机器视觉】单目测距——运动结构恢复
ps:图是随便找的,为了凑个封面 前言 在前面对光流法进行进一步改进,希望将2D光流推广至3D场景流时,发现2D转3D过程中存在尺度歧义问题,需要补全摄像头拍摄图像中缺失的深度信息,否则解空间不收敛…...
质量体系的重要
质量体系是为确保产品、服务或过程质量满足规定要求,由相互关联的要素构成的有机整体。其核心内容可归纳为以下五个方面: 🏛️ 一、组织架构与职责 质量体系明确组织内各部门、岗位的职责与权限,形成层级清晰的管理网络…...

OPENCV形态学基础之二腐蚀
一.腐蚀的原理 (图1) 数学表达式:dst(x,y) erode(src(x,y)) min(x,y)src(xx,yy) 腐蚀也是图像形态学的基本功能之一,腐蚀跟膨胀属于反向操作,膨胀是把图像图像变大,而腐蚀就是把图像变小。腐蚀后的图像变小变暗淡。 腐蚀…...

HDFS分布式存储 zookeeper
hadoop介绍 狭义上hadoop是指apache的一款开源软件 用java语言实现开源框架,允许使用简单的变成模型跨计算机对大型集群进行分布式处理(1.海量的数据存储 2.海量数据的计算)Hadoop核心组件 hdfs(分布式文件存储系统)&a…...
纯 Java 项目(非 SpringBoot)集成 Mybatis-Plus 和 Mybatis-Plus-Join
纯 Java 项目(非 SpringBoot)集成 Mybatis-Plus 和 Mybatis-Plus-Join 1、依赖1.1、依赖版本1.2、pom.xml 2、代码2.1、SqlSession 构造器2.2、MybatisPlus代码生成器2.3、获取 config.yml 配置2.3.1、config.yml2.3.2、项目配置类 2.4、ftl 模板2.4.1、…...

MFC 抛体运动模拟:常见问题解决与界面美化
在 MFC 中开发抛体运动模拟程序时,我们常遇到 轨迹残留、无效刷新、视觉单调、物理逻辑瑕疵 等问题。本文将针对这些痛点,详细解析原因并提供解决方案,同时兼顾界面美化,让模拟效果更专业、更高效。 问题一:历史轨迹与小球残影残留 现象 小球运动后,历史位置的 “残影”…...

并发编程 - go版
1.并发编程基础概念 进程和线程 A. 进程是程序在操作系统中的一次执行过程,系统进行资源分配和调度的一个独立单位。B. 线程是进程的一个执行实体,是CPU调度和分派的基本单位,它是比进程更小的能独立运行的基本单位。C.一个进程可以创建和撤销多个线程;同一个进程中…...

Unity UGUI Button事件流程
场景结构 测试代码 public class TestBtn : MonoBehaviour {void Start(){var btn GetComponent<Button>();btn.onClick.AddListener(OnClick);}private void OnClick(){Debug.Log("666");}}当添加事件时 // 实例化一个ButtonClickedEvent的事件 [Formerl…...

淘宝扭蛋机小程序系统开发:打造互动性强的购物平台
淘宝扭蛋机小程序系统的开发,旨在打造一个互动性强的购物平台,让用户在购物的同时,能够享受到更多的乐趣和惊喜。 淘宝扭蛋机小程序系统拥有丰富的互动功能。用户可以通过虚拟摇杆操作扭蛋机,实现旋转、抽拉等动作,增…...