FlinkSql 如何实现数据去重?
摘要
很多时候flink消费上游kafka的数据是有重复的,因此有时候我们想数据在落盘之前进行去重,这在实际开发中具有广泛的应用场景,此处不说详细代码,只粘贴相应的flinksql
代码
--********************************************************************--
-- 创建临时表(只在当前sessoin生效的表称为临时表) DDL
CREATE TEMPORARY TABLE UserAttrSource ( `data` string,`kafkaMetaTimestamp` TIMESTAMP(3) METADATA FROM 'timestamp', -- kafka record携带的源数据时间戳,参考官网kafka connectorproctime as PROCTIME() -- 获取数据处理时间,这是flink内置支持的关键字
) WITH ('connector' = 'kafka','topic' = 'user_attri_ad_dirty_data','properties.bootstrap.servers' = 'kafka地址','scan.startup.mode' = 'timestamp', -- kafka扫描数据模式,参考官网kafka connector'scan.startup.timestamp-millis' ='1687305600000' , -- 2023-06-21 08:00:00'format' = 'raw' -- 意思是将kafka数据格式化为string
);-- 创建SINK 表
CREATE TEMPORARY TABLE ADB (log_date DATE,`errorType` int,appId string,`errorCode` int,`errorReason` string,`deserialization` string,`originalData` string,kafkaMetaTimestamp TIMESTAMP,data_hash string,PRIMARY KEY (`data_hash`) NOT ENFORCED
)
WITH ('connector' = 'adb3.0','url' = 'jdbc:mysql://xxxx:3306/flink_data?rewriteBatchedStatements=true','tableName' = 'usr_attr_dirty', 'userName'='username','password'='password'
);
-- 去重视图, 这是关键(json_value是flink的内置函数,data_hash是数据本身的primary key)
-- 下述语句含义是:根据data_hash字段分组,按照处理时间排序,取出最新的一条数据,其他的重复数据将被抛弃
CREATE TEMPORARY VIEW quchong ASSELECT data,kafkaMetaTimestamp FROM (SELECT *,ROW_NUMBER() OVER (PARTITION BY json_value(data,'$.data_hash') ORDER BY proctime DESC) as row_numFROM UserAttrSource)WHERE row_num = 1;-- 插入目标表
insert into ADB
select TO_DATE(DATE_FORMAT(kafkaMetaTimestamp,'yyyy-MM-dd') )AS log_date,json_value(data,'$.errorType' RETURNING INT) errorType,json_value(data,'$.appId' NULL ON EMPTY) appId,json_value(data,'$.errorCode' RETURNING INT) errorCode,json_value(data,'$.errorReason' NULL ON EMPTY) errorReason,json_value(data,'$.deserialization' NULL ON EMPTY) deserialization,json_value(data,'$.originalData') originalData,kafkaMetaTimestamp,json_value(data,'$.data_hash') data_hash
from quchong;
相关文章:
FlinkSql 如何实现数据去重?
摘要 很多时候flink消费上游kafka的数据是有重复的,因此有时候我们想数据在落盘之前进行去重,这在实际开发中具有广泛的应用场景,此处不说详细代码,只粘贴相应的flinksql 代码 --********************************************…...
机器学习概念
目录 一、人工智能、机器学习、深度学习的关系 二、什么是深度学习? 2.1 深度学习常用算法 一、人工智能、机器学习、深度学习的关系 人工智能、机器学习和深度学习的关系如下所示。 二、什么是深度学习? 深度学习( DL, Deep Learning) 是机器学习 …...
【数据结构】排序(插入、选择、交换、归并) -- 详解
一、排序的概念及其运用 1、排序的概念 排序:所谓排序,就是使一串记录,按照其中的某个或某些关键字的大小,递增或递减的排列起来的操作。 稳定性:假定在待排序的记录序列中,存在多个具有相同的关键字的记…...
游戏中的图片打包流程,免费的png打包plist工具,一款把若干资源图片拼接为一张大图的免费工具
手机游戏开发中,为了提高图片渲染性能,经常需要将小图片合并成一张大图进行渲染。如果手工来做的话就非常耗时。TexturePacker就是一款非常不错方便的处理工具。TexturePacker虽然非常优秀,但不是免费的。 对于打包流程,做游戏的…...
Springboot实现ENC加密
Springboot实现ENC加密 1、导入依赖2、配置加密秘钥(盐)3、获取并配置密文4、重启项目测试5、自定义前缀、后缀6、自定义加密方式 1、导入依赖 关于版本,需要根据spring-boot版本,自行修改 <dependency><groupId>co…...
nginx 托管vue项目配置
server {listen 80;server_name your_domain.com;location / {root /path/to/your/vue/project;index index.html;try_files $uri $uri/ /index.html;} }奇怪的现象,在vue路由中/会跳转到/abc/def,但如果直接输入/abc/def会显示404,添加 try_files $uri…...
Vue3中如何进行封装?—组件之间的传值
用了很久一段时间Vue3Ts了,工作中对一些常用的组件也进行了一些封装,这里对封装的一些方法进行一些简单的总结。 1.props传递 首先在主组件进行定义传值 <template><div>这里是主组件<common :first"first"></common&…...
实训笔记8.25
实训笔记8.25 8.25笔记一、Flume数据采集技术1.1 Flume实现数据采集主要借助Flume的组成架构1.2 Flume采集数据的时候,核心是编写Flume的采集脚本xxx.conf1.2.1 脚本文件主要由五部分组成 二、Flume案例实操2.1 采集一个网络端口的数据到控制台2.1.1 分析案例的组件…...
vue自定义监听元素宽高指令
在 main.js 中添加 // 自定义监听元素高度变化指令 const resizerMap new WeakMap() const resizeObserver new ResizeObserver((entries) > {for (const entry of entries) {const handle resizerMap.get(entry.target)if (handle) {handle({width: entry.borderBoxSiz…...
网络爬虫到底是个啥?
网络爬虫到底是个啥? 当涉及到网络爬虫技术时,需要考虑多个方面,从网页获取到最终的数据处理和分析,每个阶段都有不同的算法和策略。以下是这些方面的详细解释: 网页获取(Web Crawling)&#x…...
前端行级元素和块级元素的基本区别
块级元素和行内元素的基本区别是, 行内元素可以与其他行内元素并排;块级元素独占一行,不能与其他任何元素并列; 下面看一下; <!DOCTYPE html> <html> <head> <meta charset"utf-8"&…...
CentOS 7用二进制安装MySQL5.7
[rootlocalhost ~]# [rootlocalhost ~]# ll 总用量 662116 -rw-------. 1 root root 1401 8月 29 19:29 anaconda-ks.cfg -rw-r--r--. 1 root root 678001736 8月 29 19:44 mysql-5.7.40-linux-glibc2.12-x86_64.tar.gz [rootlocalhost ~]# tar xf mysql-5.7.40-linux-…...
华为加速回归Mate 60发布, 7nm全自研工艺芯片
华为于今天12:08推出“HUAWEI Mate 60 Pro先锋计划”,让部分消费者提前体验。在华为商城看到,华为Mate 60 pro手机已上架,售价6999元,提供雅川青、白沙银、南糯紫、雅丹黑四种配色供选择。 据介绍,华为在卫星通信领域…...
Linux系列讲解 —— 【systemd】下载及编译记录
Ubuntu18.04的init程序合并到了systemd中,本篇文章记录一下systemd的下载和编译。 1. 下载systemd源码 (1) 查看systemd版本号,用来确定需要下载的分支 sunsun-pc:~$ systemd --version systemd 237 PAM AUDIT SELINUX IMA APPARMOR SMACK SYSVINIT UT…...
u-view 的u-calendar 组件设置默认日期后,多次点击后,就不滚动到默认日期的位置
场景:uniapp开发微信小程序 vue2 uview版本:2.0.36 ; u-calendar 组件设置默认日期后 我打开弹窗,再关闭弹窗, 重复两次 就不显示默认日期了 在源码中找到这个位置进行打印值,根据出bug前后的值进行…...
vue naive ui 按钮绑定按键
使用vue (naive ui) 绑定Enter 按键 知识点: 按键绑定Button全局挂载使得message,notification, dialog, loadingBar 等NaiveUI 生效UMD方式使用vue 与 naive ui将vue默认的 分隔符大括号 替换 为 [[ ]] <!DOCTYPE html> <html lang"en"> <head>…...
Viobot基本功能使用及介绍
设备拿到手当然是要先试一下效果的,这部分可以参考本专栏的第一篇 Viobot开机指南。 接下来我们就从UI开始熟悉这个产品吧! 1.状态 设备上电会自动运行它的程序,开启了一个服务器,上位机通过连接这个服务器连接到设备,…...
《PMBOK指南》第七版12大原则和8大绩效域
《PMBOK指南》第七版12大原则 原则1:成为勤勉、尊重和关心他人的管家 原则2:营造协作的项目团队环境 原则3:有效地干系人参与 原则4:聚焦于价值 原则5:识别、评估和响应系统交互 原则6:展现领导力行为…...
docker 启动命令
cd /ycw/docker docker build -f DockerFile -t jshepr:1.0 . #前面测试docker已经介绍过该命令下面就不再介绍了 docker images docker run -it -p 7003:9999 --name yyy -d jshepr:1.0 #上面运行报错 用这个 不报错就不用 docker rm yyy docker ps #查看项目日志 docker …...
C++ DAY7
一、类模板 建立一个通用的类,其类中的类型不确定,用一个虚拟类型替代 template<typename T> 类template ----->表示开始创建模板 typename -->表明后面的符号是数据类型,typename 也可以用class代替 T ----->表示数据类型…...
Vue2 使用插件 Volar 报错:<template v-for> key should be placed on the <template> tag.
目录 问题描述 版本描述 问题定位 问题解决 VS Code 插件地址 问题描述 在 VS Code 上使用插件 Volar 开发 Vue3 项目,然后去改 Vue2 项目时,对没有放在<template v-for> 元素上的 :key,会提示 <template v-for> key should…...
启动线程方法 start ()和 run ()有什么区别
在Java中,线程可以通过调用start()方法或者直接调用run()方法来执行。这两种方式有着重要的区别: start() 方法:当你调用线程的start()方法时,它会使线程进入就绪状态,等待系统调度。系统会为该线程分配资源,并在合适的时机执行线程的run()方法。实际上,start()方法会启…...
Java的全排列模板
c有全排列函数,我们Java没有,所以我们只能自己手写了。 模板一:(不去重) import java.util.ArrayList; import java.util.Random; import java.util.Scanner; public class liyunpeng {public static void main(Stri…...
读书笔记——《万物有灵》
前言 上一本书是《走出荒野》,太平洋步道女王提到了这本书《万物有灵》,她同样是看一点撕一点的阅读。我想,在她穿越山河森林,听见鸟鸣溪流的旅行过程中,是不是看这本描写动物有如何聪明的书——《万物有灵》…...
面试现场表现:展示你的编程能力和沟通技巧
🌷🍁 博主猫头虎 带您 Go to New World.✨🍁 🦄 博客首页——猫头虎的博客🎐 🐳《面试题大全专栏》 文章图文并茂🦕生动形象🦖简单易学!欢迎大家来踩踩~🌺 &a…...
34亿的mysql表如何优雅的扩字段长度兵并归档重建
业务背景: 该系统有一张表数据量已达到34亿,并且有个字段长度不够,导致很多数据无法插入。因为业务只要保留近2个月数据即可,所以需要接下来需要做2点:1,扩字段长度 2,只保留近2个月的数据。 …...
C#_进程单例模式.秒懂Mutex
什么是Mutex? 可以定义调用线程是否具有互斥性,程序创建者拥有控制权,相反只能引用程序。 参数1:如果是程序创建者,就获得控制权。 参数2:名称,可使用GUID生成。 参数3:out 返回值…...
AcWing 5050. 排序 (每日一题)
题目描述 给定一个长度为 n 的由小写字母构成的字符串。 请你按照 a∼z 的顺序,对字符串内的字符进行重新排序,并输出重新排序后的字符串。 输入格式 第一行包含整数 T ,表示共有 T 组测试数据。 每组数据第一行包含整数 n 。 第二行包…...
【TypeScript】proxy 和 Reflect
proxy (代理) 和 Reflect (反射) 参数一样。 基本用法 let person { name: xx, age: 20} // proxy 支持对象,数组,函数,set,map // 代理对象的常用模式:前面一个对象&…...
STM32f103入门(5)定时器中断
STM32 TIM(定时器/计数器)模块的中断流程如下: 配置TIM寄存器:首先,通过配置TIM相关的寄存器来设置计时器的基本参数,例如预分频系数、计数模式、计数器周期等。 使能TIM中断:使用TIM_ITConfig函…...
Mybatis查询数据
上一篇我们介绍了在pom文件中引入mybatis依赖,配置了mybatis配置文件,通过读取配置文件创建了会话工厂,使用会话工厂创建会话获取连接对象读取到了数据库的基本信息。 如果您需要对上面的内容进行了解,可以参考Mybatis引入与使用…...
【前车之鉴】: 2023最新教程-将java程序打包到maven私服的正确打开方式,详细流程介绍不怕你掌握不了
文章目录 为什么看这篇整体流程1. 注册账号【首次需要】2. 工单申请【新项目必须】3. 项目配置【新项目必须】4. 授权认证【新项目必须】5. 一键发布 最后也很重要 为什么看这篇 一是当前网络上一些博客有遗漏部分,这里做补充,二是网上思路没错ÿ…...
如何理解IaaS、PaaS、SaaS?盘点受欢迎的八大SaaS平台!
如何理解IaaS、PaaS、SaaS?他们各自的优势在哪里?我们又应该如何选择?本篇将带大家深入浅出的了解IaaS、PaaS、SaaS,并为大家盘点2023最受欢迎的SaaS平台。 通过本文你将了解: IaaS、PaaS、SaaS分别是什么我们该如何理…...
立创EDA专业版的原理图上器件有一个虚线框
立创EDA专业版的原理图上器件有一个虚线框解决方法 问题分析: 在使用立创EDA专业版 设计电路原理图时,中途莫名其妙就给我的元件添加了下面图片所示的虚线外框。看着就很别扭的样子,而且工程大了和器件稍微布局比较密的时候就导致整体很难看…...
JUC并发编程--------基础篇
一、多线程的相关知识 栈与栈帧 我们都知道 JVM 中由堆、栈、方法区所组成,其中栈内存是给谁用的呢?其实就是线程,每个线程启动后,虚拟 机就会为其分配一块栈内存。 每个栈由多个栈帧(Frame)组成…...
秒懂算法2
视频链接 : 希望下次秒懂的是算法题_哔哩哔哩_bilibili P1094 [NOIP2007 普及组] 纪念品分组 原题链接 : [NOIP2007 普及组] 纪念品分组 - 洛谷 思路 : 排序 贪心 双指针首先先对输入进来的数组进行排序(由小到大)运用贪心的思想 : 前后结合,令l1,rn,若a[l]a[r]<w…...
隐秘的角落:Java连接Oracle提示Connection timed out
前言 这个报错相信各位后端开发都不陌生,大体的原因就那么几种: 检查网络连接:确保您的计算机与数据库服务器之间的网络连接正常。尝试通过其他方式验证您的网络连接是否正常。 检查数据库服务器状态:确保数据库服务器正在运行&…...
基于微信小程序的餐厅预订系统的设计与实现(论文+源码)_kaic
摘 要 随着消费升级,越来越多的年轻人已经开始不再看重餐饮等行业的服务,而是追求一种轻松自在的用餐、购物环境。因此,无人餐厅、无人便利店、无人超市等一些科技消费场所应势而生。餐饮企业用工荒已成为不争的事实。服务员行业的低保障、低…...
科技政策 | 四川省科学技术厅关于发布2024年第一批省级科技计划项目申报指南的通知
原创 | 文 BFT机器人 近日,四川省科学技术厅发布了2024年第一批省级科技计划项目申报指南;其中包括自然科学基金项目、重点研发计划、科技成果转移转化引导计划、科技创新基地(平台)和人才计划。 01 自然科学基金项目 实施周期 …...
深入了解Webpack:特性、特点和结合JS混淆加密的实例
Webpack是现代前端开发中最受欢迎的构建工具之一,其强大的特性和灵活性使得开发者能够更有效地管理和优化项目资源。在本文中,我们将深入探讨Webpack的特性和特点,并结合实例演示如何使用Webpack与JS混淆加密相结合。Webpack的特性和特点 1.…...
2023-08-23力扣每日一题
链接: 1782. 统计点对的数目 题意: 给n个点和m条无向边(可重复),q个查询 定义edge[a]为一个点是a的边数量,定义ret[a,b]是edge[a]edge[b]-(a与b的边) q个查询q个答案࿰…...
分发饼干【贪心算法】
分发饼干 假设你是一位很棒的家长,想要给你的孩子们一些小饼干。但是,每个孩子最多只能给一块饼干。 对每个孩子 i,都有一个胃口值 g[i],这是能让孩子们满足胃口的饼干的最小尺寸;并且每块饼干 j,都有一个…...
为什么网络互联地址设置为30位地址
对于点对点链路,为了节约IPv4地址,一般为其分配/30地址块,这样包含4个地址:最小地址作为网络地址,最大地址作为广播地址,剩余两个可分配地址,分配给链路两端的接口,这是最普遍的方法…...
青少年棒球锦标赛发展·棒球1号位
青少年棒球锦标赛发展 1. 青少年棒球锦标赛简介 青少年棒球锦标赛是一个令人兴奋的国际性比赛,每年都有来自世界各地的优秀青少年棒球选手参加。这个锦标赛旨在提供一个展示青少年棒球选手的技能和才华的平台,同时也是为了推动棒球在全球范围内的普及和…...
Unity实现UI图片面板滚动播放效果第二弹
效果: 场景结构: 特殊物体:panel下面用排列组件horizent layout group放置多个需要显示的面板,用mask遮罩好。 主要思路: 这次是要在最后一个toggle的地方,依然向左滚动回1,这是难点。因此实际…...
Redis的基本操作
文章目录 1.Redis简介2.Redis的常用数据类型3.Redis的常用命令1.字符串操作命令2.哈希操作命令3.列表操作命令4.集合操作命令5.有序集合操作命令6.通用操作命令 4.Springboot配置Redis1.导入SpringDataRedis的Maven坐标2.配置Redis的数据源3.编写配置类,创还能Redis…...
省级智慧农业大数据平台项目规划建设方案[195页Word]
导读:原文《省级智慧农业大数据平台项目规划建设方案[195页Word]》(获取来源见文尾),本文精选其中精华及架构部分,逻辑清晰、内容完整,为快速形成售前方案提供参考。 1 农业大数据平台项目概述 1.1 建设…...
php图片批量压缩并同时保持清晰度
php图片压缩可以通过GD库来实现。以下是一个使用GD库进行图片压缩的示例代码: // 原始图片路径 $sourceImage path/to/source/image.jpg; // 压缩后保存的路径及文件名 $compressedImage path/to/compressed/image.jpg; // 压缩后的图片质量(1-100&…...
243:vue+Openlayers 更改鼠标滚轮缩放地图大小,每次缩放小一点
第243个 点击查看专栏目录 本示例的目的是介绍如何在vue+openlayers项目中设置鼠标滚轮缩放地图大小,每次滑动一格滚轮,设定的值非默认值1。具体的设置方法,参考源代码。 直接复制下面的 vue+openlayers源代码,操作2分钟即可运行实现效果 文章目录 示例效果配置方式示例源…...
NOI2015D. 荷马史诗
荷马史诗 题目描述 追逐影子的人,自己就是影子。 ——荷马 Allison 最近迷上了文学。她喜欢在一个慵懒的午后,细细地品上一杯卡布奇诺,静静地阅读她爱不释手的《荷马史诗》。但是由《奥德赛》和《伊利亚特》组成的鸿篇巨制《荷马史诗》实在是…...