Flink SQL 功能解密系列 —— 数据去重的技巧和思考

本文涉及的产品
实时计算 Flink 版,5000CU*H 3个月
简介: 去重逻辑在业务处理中使用广泛,大致可以分两类:DISTINCT去重和FIRST_VALUE主键去重,两者的区别是DISTINCT去重是对整行数据进行去重,比如tt里面数据可能会有重复,我们要去掉重复的数据;FIRST_VALUE是根据主键进行去重,可以看成是一种业务层面的去重,但是真实的业务场景使用也很普遍,比如一个用户有多次点击,业务上只需要取第一条。

概述

去重逻辑在业务处理中使用广泛,大致可以分两类:DISTINCT去重和FIRST_VALUE主键去重,两者的区别是DISTINCT去重是对整行数据进行去重,比如tt里面数据可能会有重复,我们要去掉重复的数据;FIRST_VALUE是根据主键进行去重,可以看成是一种业务层面的去重,但是真实的业务场景使用也很普遍,比如一个用户有多次点击,业务上只需要取第一条。本文重点介绍这两种去重的应用。

1. DISTINCT 去重

blink sql支持标准sql的DISTINCT去重。假如我们有如下输入数据,并希望对相同的行进行去重。

image.png

sql可以这么写:select distinct * from tt_source; 完整的blink sql如下,

create table tt_source(
  a varchar,
  b varchar
)with(
  type='tt',
  topic='se_taobao_wireless_click',
  accessId='08061416466YCN3FIU',
  accessKey='xxxxx'
  lengthCheck='PAD'
);


create table tt_output(
  a varchar,
  b varchar
)with(
  type='tt',
  topic='blink_test_32_1',
  accessKey='xxxx'
);

insert into tt_output
select distinct * from tt_source;
AI 代码解读

输出时,会对第一行(1,1)和第二行(1,1)数据进行去重。输出结果如下

image.png

2. FIRST_VALUE udaf去重

还有一种情况是根据primary key字段进行去重,即如果两行数据主键相同,即使其他非主键字段不一样,还是只取第一行数据。这种情况,我们可以使用FIRST_VALUE udaf函数来达到去重的目的。
对于如下输入,并希望根据主键a来去重数据:

image.png

sql可以这么写:

INSERT INTO tt_output
SELECT 
  a, 
  FIRST_VALUE(b) 
FROM tt_source
GROUP BY a;
AI 代码解读

完整的blink sql如下,

CREATE TABLE tt_source(
  a VARCHAR,
  b VARCHAR
)WITH(
  type='tt',
  topic='se_taobao_wireless_click',
  accessId='08061416466YCN3FIU',
  accessKey='xxx',
  lengthCheck='PAD'
);

CREATE TABLE tt_output(
  a VARCHAR,
  b VARCHAR
)WITH(
  type='tt',
  topic='blink_test_32_1',
  accessKey='xxx'
);

INSERT INTO tt_output
SELECT 
  a, 
  FIRST_VALUE(b) 
FROM tt_source
GROUP BY a;
AI 代码解读

输出结果:

image.png

可以看到主键a相同的3行,只取了第一行。

FIRST_VALUE还支持传一个order参数,根据order来决定first是哪行,使用的方法是FIRST_VALUE(b, c),但是要注意,c字段只能是BIGINT。假如我们有如下输入,对于相同的主键,我们希望取c最小的记录(实际场景c一般是时间字段)。

image.png

完整的blink sql如下,

CREATE TABLE tt_source(
  a VARCHAR,
  b VARCHAR,
  c BIGINT
)WITH(
  type='tt',
  topic='se_taobao_wireless_click',
  accessId='08061416466YCN3FIU',
  accessKey='xxx',
  lengthCheck='PAD'
);

CREATE TABLE tt_output(
  a VARCHAR,
  b VARCHAR
)WITH(
  type='tt',
  topic='blink_test_32_1',
  accessKey='xxx'
);

INSERT INTO tt_output
SELECT 
  a, 
  FIRST_VALUE(b, c) 
FROM tt_source
GROUP BY a;
AI 代码解读

输出结果:

image.png

可以看到当输出(1,1,1)后,由于又来了(1,2,0),0比1要小,所以又更新了主键为1的记录,输出(1,2)

相关实践学习
基于Hologres+Flink搭建GitHub实时数据大屏
通过使用Flink、Hologres构建实时数仓,并通过Hologres对接BI分析工具(以DataV为例),实现海量数据实时分析.
Linux入门到精通
本套课程是从入门开始的Linux学习课程,适合初学者阅读。由浅入深案例丰富,通俗易懂。主要涉及基础的系统操作以及工作中常用的各种服务软件的应用、部署和优化。即使是零基础的学员,只要能够坚持把所有章节都学完,也一定会受益匪浅。
目录
打赏
0
0
0
1
36037
分享
相关文章
SQL如何在CTE中使用Order By的功能
SQL Server如何在CTE中使用Order By的功能
基于 Flink 的中国电信星海时空数据多引擎实时改造
本文整理自中国电信集团大数据架构师李新虎老师在Flink Forward Asia 2024的分享,围绕星海时空智能系统展开,涵盖四个核心部分:时空数据现状、实时场景多引擎化、典型应用及未来展望。系统日处理8000亿条数据,具备亚米级定位能力,通过Flink多引擎架构解决数据膨胀与响应时效等问题,优化资源利用并提升计算效率。应用场景包括运动状态识别、个体行为分析和群智感知,未来将推进湖仓一体改造与三维时空服务体系建设,助力数字化转型与智慧城市建设。
553 3
基于 Flink 的中国电信星海时空数据多引擎实时改造
京东零售基于Flink的推荐系统智能数据体系
摘要:本文整理自京东零售技术专家张颖老师,在 Flink Forward Asia 2024 生产实践(二)专场中的分享,介绍了基于Flink构建的推荐系统数据,以及Flink智能体系带来的智能服务功能。内容分为以下六个部分: 推荐系统架构 索引 样本 特征 可解释 指标 Tips:关注「公众号」回复 FFA 2024 查看会后资料~
133 1
京东零售基于Flink的推荐系统智能数据体系
【YashanDB知识库】Flink CDC实时同步Oracle数据到崖山
本文介绍通过Flink CDC实现Oracle数据实时同步至崖山数据库(YashanDB)的方法,支持全量与增量同步,并涵盖新增、修改和删除的DML操作。内容包括环境准备(如JDK、Flink版本等)、Oracle日志归档启用、用户权限配置、增量日志记录设置、元数据迁移、Flink安装与配置、生成Flink SQL文件、Streampark部署,以及创建和启动实时同步任务的具体步骤。适合需要跨数据库实时同步方案的技术人员参考。
【YashanDB知识库】Flink CDC实时同步Oracle数据到崖山
SpringBoot 通过集成 Flink CDC 来实时追踪 MySql 数据变动
通过详细的步骤和示例代码,您可以在 SpringBoot 项目中成功集成 Flink CDC,并实时追踪 MySQL 数据库的变动。
1389 43
Flink SQL 详解:流批一体处理的强大工具
Flink SQL 是为应对传统数据处理框架中流批分离的问题而诞生的,它融合了SQL的简洁性和Flink的强大流批处理能力,降低了大数据处理门槛。其核心工作原理包括生成逻辑执行计划、查询优化和构建算子树,确保高效执行。Flink SQL 支持过滤、投影、聚合、连接和窗口等常用算子,实现了流批一体处理,极大提高了开发效率和代码复用性。通过统一的API和语法,Flink SQL 能够灵活应对实时和离线数据分析场景,为企业提供强大的数据处理能力。
1051 26
|
8月前
|
Flink SQL Deduplication 去重以及如何获取最新状态操作
Flink SQL Deduplication 是一种高效的数据去重功能,支持多种数据类型和灵活的配置选项。它通过哈希表、时间窗口和状态管理等技术实现去重,适用于流处理和批处理场景。本文介绍了其特性、原理、实际案例及源码分析,帮助读者更好地理解和应用这一功能。
571 14
5-10Can't connect to MySQL server on 'sh-cynosl-grp-fcs50xoa.sql.tencentcdb.com' (110)")
5-10Can't connect to MySQL server on 'sh-cynosl-grp-fcs50xoa.sql.tencentcdb.com' (110)")

相关产品

  • 实时计算 Flink版
  • AI助理

    你好,我是AI助理

    可以解答问题、推荐解决方案等