当前位置: 首页 > news >正文

大数据-玩转数据-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 的架构 总结以下几种方案实现定时任务&#xff0c;可根据不同需求去使用不同方案。 while True: sleep() 利用whil…...

AcWing算法提高课-5.6.1同余方程

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

Docker Tutorial

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

平面图—简单应用

平面图&#xff1a;若一个图&#x1d43a;能画在平面&#x1d446;上&#xff0c;且使&#x1d43a;的边仅在端点处相交&#xff0c;则称图&#x1d43a;为可嵌入平面&#x1d446;&#xff0c;&#x1d43a;称为可平面图&#xff0c;简称为平面图。 欧拉公式&#xff1a;设有…...

安装JDK(Java SE Development Kit)超详细教程

文章时间 &#xff1a; 2023-10-04 1. 下载地址 直接去下载地址&#xff1a;https://www.oracle.com/java/technologies/downloads/ &#xff08;需要翻墙&#xff0c;不想翻墙或者不想注册oracel账号的&#xff0c;直接去我的阿里云盘&#xff09; 阿里云盘&#xff1a;http…...

KUKA机器人通过3点法设置工作台基坐标系的具体方法

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

以太网的MAC层

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

Hadoop启动后jps发现没有DateNode解决办法

多次使用 Hadoop namenode -format 格式化节点后DateNode丢失 找到hadoop配置文件core-site.xml查找tmp路径 进入该路径&#xff0c;使用rm -rf data删除data文件 再次使用Hadoop namenode -format 格式化后jps后出现DateNode节点...

VUE3照本宣科——应用实例API与setup

VUE3照本宣科——应用实例API与setup 前言一、应用实例API1.createApp()2.app.use()3.app.mount() 二、setup 前言 &#x1f468;‍&#x1f4bb;&#x1f468;‍&#x1f33e;&#x1f4dd;记录学习成果&#xff0c;以便温故而知新 “VUE3照本宣科”是指照着中文官网和菜鸟教…...

json/js对象的key有什么区别?

1.对于JS对象来说 一个js对象如果是这样的 obj {"0": "小明","0name": "小明明", "": 18,"&#xffe5;": "哈哈"," ": "爱好广泛" }对于js对象来说&#xff0c;有时候key是不…...

极大似然估计概念的理解——统计学习方法

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

python模拟表格任意输入位置

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

如何限制文件只能通过USB打印机打印,限制打印次数和时限并且无法在打印前查看或编辑内容

在今天这个高度信息化的时代&#xff0c;文档打印已经成为日常工作中不可或缺的一部分。然而&#xff0c;这也带来了诸多安全风险&#xff0c;如文档被篡改、知识产权被侵犯以及信息泄露等。为了解决这些问题&#xff0c;只印应运而生。作为一款独特的软件工具&#xff0c;只印…...

车牌文本检测与识别:License Plate Recognition Based On Multi-Angle View Model

论文作者&#xff1a;Dat Tran-Anh,Khanh Linh Tran,Hoai-Nam Vu 作者单位&#xff1a;Thuyloi University;Posts and Telecommunications Institute of Technology 论文链接&#xff1a;http://arxiv.org/abs/2309.12972v1 内容简介&#xff1a; 1&#xff09;方向&#x…...

Blender中的4种视图着色模式

Blender中有四种主要的视图着色模式&#xff1a;线框、实体、Look Dev和渲染。它们的主要区别如下&#xff1a; - 线框模式只显示物体的边缘&#xff08;线框&#xff09;&#xff0c;可以让您看到场景中的所有物体&#xff0c;也可以调整线框的颜色和背景的颜色。 - 实…...

Flutter项目安装到Android手机一直显示在assembledebug

问题 Flutter项目安装到Android手机一直显示在assembledebug 原因 网络不好&#xff0c;gradle依赖下载不下来 解决方案 修改如下的文件 gradle-wrapper.properties 使用腾讯提供的gradle镜像下载 distributionUrlhttps://mirrors.cloud.tencent.com/gradle/gradle-7.5…...

数据挖掘实验(二)数据预处理【等深分箱与等宽分箱】

一、分箱平滑的原理 &#xff08;1&#xff09;分箱方法 在分箱前&#xff0c;一定要先排序数据&#xff0c;再将它们分到等深&#xff08;等宽&#xff09;的箱中。 常见的有两种分箱方法&#xff1a;等深分箱和等宽分箱。 等深分箱&#xff1a;按记录数进行分箱&#xff0…...

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 我这边…...

stm32G473的flash模式是单bank还是双bank?

今天突然有人stm32G473的flash模式是单bank还是双bank&#xff1f;由于时间太久&#xff0c;我真忘记了。搜搜发现&#xff0c;还真有人和我一样。见下面的链接&#xff1a;https://shequ.stmicroelectronics.cn/forum.php?modviewthread&tid644563 根据STM32G4系列参考手…...

Golang dig框架与GraphQL的完美结合

将 Go 的 Dig 依赖注入框架与 GraphQL 结合使用&#xff0c;可以显著提升应用程序的可维护性、可测试性以及灵活性。 Dig 是一个强大的依赖注入容器&#xff0c;能够帮助开发者更好地管理复杂的依赖关系&#xff0c;而 GraphQL 则是一种用于 API 的查询语言&#xff0c;能够提…...

全球首个30米分辨率湿地数据集(2000—2022)

数据简介 今天我们分享的数据是全球30米分辨率湿地数据集&#xff0c;包含8种湿地亚类&#xff0c;该数据以0.5X0.5的瓦片存储&#xff0c;我们整理了所有属于中国的瓦片名称与其对应省份&#xff0c;方便大家研究使用。 该数据集作为全球首个30米分辨率、覆盖2000–2022年时间…...

智能在线客服平台:数字化时代企业连接用户的 AI 中枢

随着互联网技术的飞速发展&#xff0c;消费者期望能够随时随地与企业进行交流。在线客服平台作为连接企业与客户的重要桥梁&#xff0c;不仅优化了客户体验&#xff0c;还提升了企业的服务效率和市场竞争力。本文将探讨在线客服平台的重要性、技术进展、实际应用&#xff0c;并…...

Java - Mysql数据类型对应

Mysql数据类型java数据类型备注整型INT/INTEGERint / java.lang.Integer–BIGINTlong/java.lang.Long–––浮点型FLOATfloat/java.lang.FloatDOUBLEdouble/java.lang.Double–DECIMAL/NUMERICjava.math.BigDecimal字符串型CHARjava.lang.String固定长度字符串VARCHARjava.lang…...

视频字幕质量评估的大规模细粒度基准

大家读完觉得有帮助记得关注和点赞&#xff01;&#xff01;&#xff01; 摘要 视频字幕在文本到视频生成任务中起着至关重要的作用&#xff0c;因为它们的质量直接影响所生成视频的语义连贯性和视觉保真度。尽管大型视觉-语言模型&#xff08;VLMs&#xff09;在字幕生成方面…...

用docker来安装部署freeswitch记录

今天刚才测试一个callcenter的项目&#xff0c;所以尝试安装freeswitch 1、使用轩辕镜像 - 中国开发者首选的专业 Docker 镜像加速服务平台 编辑下面/etc/docker/daemon.json文件为 {"registry-mirrors": ["https://docker.xuanyuan.me"] }同时可以进入轩…...

如何理解 IP 数据报中的 TTL?

目录 前言理解 前言 面试灵魂一问&#xff1a;说说对 IP 数据报中 TTL 的理解&#xff1f;我们都知道&#xff0c;IP 数据报由首部和数据两部分组成&#xff0c;首部又分为两部分&#xff1a;固定部分和可变部分&#xff0c;共占 20 字节&#xff0c;而即将讨论的 TTL 就位于首…...

基于matlab策略迭代和值迭代法的动态规划

经典的基于策略迭代和值迭代法的动态规划matlab代码&#xff0c;实现机器人的最优运输 Dynamic-Programming-master/Environment.pdf , 104724 Dynamic-Programming-master/README.md , 506 Dynamic-Programming-master/generalizedPolicyIteration.m , 1970 Dynamic-Programm…...

【从零学习JVM|第三篇】类的生命周期(高频面试题)

前言&#xff1a; 在Java编程中&#xff0c;类的生命周期是指类从被加载到内存中开始&#xff0c;到被卸载出内存为止的整个过程。了解类的生命周期对于理解Java程序的运行机制以及性能优化非常重要。本文会深入探寻类的生命周期&#xff0c;让读者对此有深刻印象。 目录 ​…...