当前位置: 首页 > 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 我这边…...

在鸿蒙HarmonyOS 5中实现抖音风格的点赞功能

下面我将详细介绍如何使用HarmonyOS SDK在HarmonyOS 5中实现类似抖音的点赞功能&#xff0c;包括动画效果、数据同步和交互优化。 1. 基础点赞功能实现 1.1 创建数据模型 // VideoModel.ets export class VideoModel {id: string "";title: string ""…...

【机器视觉】单目测距——运动结构恢复

ps&#xff1a;图是随便找的&#xff0c;为了凑个封面 前言 在前面对光流法进行进一步改进&#xff0c;希望将2D光流推广至3D场景流时&#xff0c;发现2D转3D过程中存在尺度歧义问题&#xff0c;需要补全摄像头拍摄图像中缺失的深度信息&#xff0c;否则解空间不收敛&#xf…...

质量体系的重要

质量体系是为确保产品、服务或过程质量满足规定要求&#xff0c;由相互关联的要素构成的有机整体。其核心内容可归纳为以下五个方面&#xff1a; &#x1f3db;️ 一、组织架构与职责 质量体系明确组织内各部门、岗位的职责与权限&#xff0c;形成层级清晰的管理网络&#xf…...

OPENCV形态学基础之二腐蚀

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

HDFS分布式存储 zookeeper

hadoop介绍 狭义上hadoop是指apache的一款开源软件 用java语言实现开源框架&#xff0c;允许使用简单的变成模型跨计算机对大型集群进行分布式处理&#xff08;1.海量的数据存储 2.海量数据的计算&#xff09;Hadoop核心组件 hdfs&#xff08;分布式文件存储系统&#xff09;&a…...

纯 Java 项目(非 SpringBoot)集成 Mybatis-Plus 和 Mybatis-Plus-Join

纯 Java 项目&#xff08;非 SpringBoot&#xff09;集成 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. 进程是程序在操作系统中的一次执行过程&#xff0c;系统进行资源分配和调度的一个独立单位。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…...

淘宝扭蛋机小程序系统开发:打造互动性强的购物平台

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