Flink SQL 知其所以然：Deduplication去重以及如何获取最新状态操作
发布时间：2025-11-05 13:00:20 来源：创站工坊 作者：IT科技
大家好，知其重及作我是去取最老羊，今天我们来学习 Flink SQL 中的何获 Deduplication 去重以及如何通过 Deduplication 操作获取最新的状态。
Deduplication 定义（支持 Batch\Streaming）：Deduplication 其实就是新状去重，也即上文介绍到的态操 TopN 中 row_number = 1 的场景，但是知其重及作这里有一点不一样在于其排序字段一定是时间属性列，不能是去取最其他非时间属性的普通列。在 row_number = 1 时，何获如果排序字段是新状普通列 planner 会翻译成 TopN 算子，如果是态操时间属性列 planner 会翻译成 Deduplication，这两者最终的知其重及作执行算子是不一样的，Deduplication 相比 TopN 算子专门做了对应的去取最优化，性能会有很大提升。何获应用场景：比如上游数据发重了，新状或者计算 DAU 明细数据等场景，态操都可以使用 Deduplication 语法去做去重。SQL 语法标准： 复制SELECT [column_list]FROM ( SELECT [column_list], ROW_NUMBER() OVER ([PARTITION BY col1[, col2...]] ORDER BY time_attr [asc|desc]) ASrownum
FROM table_name)WHERE rownum = 11.2.3.4.5.6.7.其中：
ROW_NUMBER()：标识当前数据的排序值。服务器租用PARTITION BY col1[, col2...]：标识分区字段，代表按照这个 col 字段作为分区粒度对数据进行排序。ORDER BY time_attr [asc|desc]：标识排序规则，必须为时间戳列，当前 Flink SQL 支持处理时间、事件时间，ASC 代表保留第一行，DESC 代表保留最后一行。WHERE rownum = 1：这个子句是一定需要的，而且必须为 rownum = 1。 实际案例：博主这里举两个案例：
案例 1（事件时间）：是腾讯 QQ 用户等级的场景，每一个 QQ 用户都有一个 QQ 用户等级，需要求出当前用户等级在星星，月亮，太阳 的用户数分别有多少。 复制-- 数据源：当每一个用户的等级初始化及后续变化的时候的数据，即用户等级变化明细数据。CREATE TABLE source_table ( user_id BIGINT COMMENT 用户 id, level STRING COMMENT 用户等级, row_time AS cast(CURRENT_TIMESTAMP as timestamp(3)) COMMENT 事件时间戳, WATERMARK FOR row_time ASrow_time
) WITH ( connector = datagen, rows-per-second = 1, fields.level.length = 1, fields.user_id.min = 1, fields.user_id.max = 1000000);-- 数据汇：输出即每一个等级的用户数CREATE TABLE sink_table ( level STRING COMMENT 等级, uv BIGINT COMMENT 当前等级用户数, row_time timestamp(3) COMMENT 时间戳) WITH ( connector = print);-- 处理逻辑：INSERT INTOsink_table
selectlevel
, count(1) asuv
, max(row_time) asrow_time
from ( SELECT user_id, level, row_time, row_number() over(partition by user_id order by row_time) asrn
FROMsource_table
)where rn = 1group by level1.2.3.4.5.6.7.8.9.10.11.12.13.14.15.16.17.18.19.20.21.22.23.24.25.26.27.28.29.30.31.32.33.34.35.36.37.38.输出结果：
复制+I[等级 1, 6928, 2021-1-28T22:34]-I[等级 1, 6928, 2021-1-28T22:34]+I[等级 1, 8670, 2021-1-28T22:34]-I[等级 1, 8670, 2021-1-28T22:34]+I[等级 1, 77287, 2021-1-28T22:34]...1.2.3.4.5.6.可以看到其有回撤数据。
其对应的 SQL 语义如下：
数据源：消费到 Kafka 中数据后，将数据按照 partition by 的 key 通过 hash 分发策略发送到下游去重算子。Deduplication 去重算子：接受到上游数据之后，根据 order by 中的站群服务器条件判断当前的这条数据和之前数据时间戳大小，以上面案例来说，如果当前数据时间戳大于之前数据时间戳，则撤回之前向下游发的中间结果，然后将最新的结果发向下游（发送策略也为 hash，具体的 hash 策略为按照 group by 中 key 进行发送），如果当前数据时间戳小于之前数据时间戳，则不做操作。次算子产出的结果就是每一个用户的对应的最新等级信息。Group by 聚合算子：接受到上游数据之后，根据 Group by 聚合粒度对数据进行聚合计算结果（每一个等级的用户数），发往下游数据汇算子。数据汇：接收到上游的数据之后，然后输出到外部存储引擎中。 案例 2（处理时间）：最原始的日志是明细数据，需要我们根据用户 id 筛选出这个用户当天的第一条数据，发往下游，下游可以据此计算分各种维度的 DAU。 复制-- 数据源：原始日志明细数据CREATE TABLE source_table ( user_id BIGINT COMMENT 用户 id, name STRING COMMENT 用户姓名, server_timestamp BIGINT COMMENT 用户访问时间戳, proctime AS PROCTIME()) WITH ( connector = datagen, rows-per-second = 1, fields.name.length = 1, fields.user_id.min = 1, fields.user_id.max = 10, fields.server_timestamp.min = 1, fields.server_timestamp.max = 100000);-- 数据汇：根据 user_id 去重的云南idc服务商第一条数据CREATE TABLE sink_table ( user_id BIGINT, name STRING, server_timestamp BIGINT) WITH ( connector = print);-- 处理逻辑：INSERT INTOsink_table
select user_id, name,server_timestamp
from ( SELECT user_id, name, server_timestamp, row_number() over(partition by user_id order by proctime) asrn
FROMsource_table
)where rn = 11.2.3.4.5.6.7.8.9.10.11.12.13.14.15.16.17.18.19.20.21.22.23.24.25.26.27.28.29.30.31.32.33.34.35.36.37.38.输出结果：
复制+I[1, 用户 1, 2021-1-28T22:34]+I[2, 用户 2, 2021-1-28T22:34]+I[3, 用户 3, 2021-1-28T22:34]...1.2.3.4.可以看到这个处理逻辑是没有回撤数据的。其对应的 SQL 语义如下：
数据源：消费到 Kafka 中数据后，将数据按照 partition by 的 key 通过 hash 分发策略发送到下游去重算子。Deduplication 去重算子：处理时间语义下，如果是当前 key 的第一条数据，则直接发往下游，如果判断（根据 state 中是否存储过改 key）不是第一条，则直接丢弃。数据汇：接收到上游的数据之后，然后输出到外部存储引擎中。注意：
在 Deduplication 关于是否会出现回撤流，博主总结如下：
⭐ Order by 事件时间 DESC：会出现回撤流，因为当前 key 下可能会有 比当前事件时间还大的数据。⭐ Order by 事件时间 ASC：会出现回撤流，因为当前 key 下可能会有 比当前事件时间还小的数据。⭐ Order by 处理时间 DESC：会出现回撤流，因为当前 key 下可能会有 比当前处理时间还大的数据。⭐ Order by 处理时间 ASC：不会出现回撤流，因为当前 key 下不可能会有 比当前处理时间还小的数据。- 上一篇：通过查看网页源代码可以发现，网页设计师们最常用的是 Windows 平台中的“微软雅黑”、“宋体”等字体。与我在经验“Ubuntu 里为 LibreOffice 设置替换显示字体”中提到的相似的原因，在 Ubuntu 中浏览网页时，由于字体对不上，网页内容一般也都是显示为黑体字，比较单调。除了从 Windows 平台复制字体到 Ubuntu 平台这一方法之外，其实我们可以通过设定字体替换，让网页上常用的“黑体”、“宋体”由 Ubuntu 调用相应的字体显示，让网页外观在两个平台上有近似的显示效果。在 Ubuntu 平台上打开网页，满眼都是黑体字。难道网页设计师就是这样设计的吗？在网页空白处右键，在弹出菜单中点击“查看源代码”（或类似的选项），在其中找到网页 css 链接并打开，可以看到，网页设计师并不是只用“黑体”一种字体。如图，该网页的设计师使用了“黑体”、“宋体”两种字体显示中文。为了让 Ubuntu 平台上浏览网页时，能够看到与网页设计师原始设计接近的效果，我们需要替换字体。软件名称：Font Creator Program v4.1 绿色汉化破解版(字体制作编辑修改工具)软件大小：3MB更新时间：2013-05-131、首先，要安装一款叫做“Font Manager“的软件。Font Manager 安装完成后，点击启动它。待启动完成后，在窗口左侧下方，点击字体设置按钮（上面带有a、b两个字母的按钮）。2、在弹出的菜单中，点选“Alias Editor“。3、弹出 Alias Editor 窗口，点击右侧的 Add Alias （添加字体别名）按钮。4、左侧输入窗口中会出现一个默认字符为“Family”的条目，双击它，并改为“宋体”，然后回车；再点击窗口右侧的 “Add Substitute”（添加字体替换）按钮。5、在上一步骤输入“宋体”的条目下方，会出现默认字符为“Subtitute”的条目，同样双击，改为“AR PL UMing CN”（输入“AR”两个字母后，下方会弹出字体列表，可以从中选择）、回车。6、用同样的方法，输入以下字体及其替换：SimSun——AR PL UMing CN；黑体——Droid Sans Fallback；等等，如图。7、最后点击窗口右下角的“Write configure”（写入配置文件）按钮，窗口关闭，回到 Font Manager 主窗口，在弹出的 Font Manager needs to reload 对话框中，点击“Reload”。8、注销 Ubuntu 登录（注意保存正在编辑的文档）、重新登录。9、再打开第1步骤中打开过的网页，会看到网页字体区分内容，由不同的字体显示不同的内容了。标题由黑体负责显示，内容由宋体负责显示，错落有致，不会视觉疲劳。相关推荐： Ubuntu系统中LibreOffice怎么替换显示字体？ ubuntu系统中删除/卸载自行安装的字体的方法
- 下一篇：诺基亚S8手机（功能全面，性能卓越，诺基亚S8是您的最佳选择）
- DostyleHS309耳机（探索音乐的终极享受，尽在DostyleHS309耳机）
- MyBatis中的十个宝藏技巧！
- 如何主动清空.Net数据库连接池？
- 什么是 SQL 注入，这些坑得避开
- 电脑端口错误代码52的解决方法（详解电脑端口错误代码52及其修复措施）
- 如何从 Linux 发行版的仓库中安装 Java
- 物联网产业104页深度研究报告：物联网研究框架与投资机会分析
- Redis大Key问题如何排查？如何解决？
- 对于Ubuntu Unity 下找回系统托盘，只需要将程序纳入白名单中即可。但是在Ubuntu 13.04 中，原本的方法行不通，因为配置文件中没有com.canonical.Unity.Panel 选项。这个时候，需要把这个选项重新激活，使用下面的PPA即可：sudo add-apt-repository ppa:timekiller/unity-systrayfixsudo apt-get updatesudo apt-get upgrade完成之后，注销或者重启。然后就可以按照以前的方法设置白名单了（下面是设置全部）：gsettings set com.canonical.Unity.Panel systray-whitelist [all]Ubuntu Unity 下无法显示QQ、Pidgin 等系统托盘的图标的解决方法这是由于在Unity中，Ubuntu对系统托盘采取了白名单机制，只有在白名单内的程序才能在系统托盘处显示图标。假如你想所有的程序都可以显示，则在终端运行下面命令：gsettings set com.canonical.Unity.Panel systray-whitelist [all]若你想自定义程序的话，运行下面的命令（YOUR_APPLICATION是程序名）：gsettings set com.canonical.Unity.Panel systray-whitelist “[JavaEmbeddedFrame, Mumble, Wine, YOUR_APPLICATION]“还原到系统默认状态：gsettings set com.canonical.Unity.Panel systray-whitelist [JavaEmbeddedFrame, Mumble, Wine, Skype, hp-systray]Ubuntu 13.04 下默认没有com.canonical.Unity.Panel 选项，需要安装：sudo add-apt-repository ppa:timekiller/unity-systrayfixsudo apt-get updatesudo apt-get upgrade成功后再执行上诉方法即可。
- 为物联网选择合适的微控制器的技巧
随便看看
IT技术网企商汇亿华云服务器租用香港云服务器源码库IT资讯网源码下载益华科技益强智未来亿华互联极客编程益强科技技术快报汇智坊益强IT技术网益强资讯优选思维库IT资讯网亿华云益强编程舍全栈开发亿华科技多维IT资讯益华科技亿华智慧云创站工坊益强智囊团运维纵横智能时代益强前沿资讯极客码头亿华智造益强编程堂科技前瞻云站无忧亿华灵动亿华云计算益强科技益华IT技术论坛码上建站益强数据堂
- Copyright © 2025 Powered by Flink SQL 知其所以然：Deduplication去重以及如何获取最新状态操作,创站工坊 滇ICP备2023006006号-44sitemap