[实战-10]FlinkSql 如何实现数据去重?

这篇具有很好参考价值的文章主要介绍了[实战-10]FlinkSql 如何实现数据去重?。希望对大家有所帮助。如果存在错误或未考虑完全的地方,请大家不吝赐教,您也可以点击"举报违法"按钮提交疑问。

摘要

很多时候flink消费上游kafka的数据是有重复的,因此有时候我们想数据在落盘之前进行去重,这在实际开发中具有广泛的应用场景,此处不说详细代码,只粘贴相应的flinksql文章来源地址https://www.toymoban.com/news/detail-776441.html

代码

--********************************************************************--
-- 创建临时表(只在当前sessoin生效的表称为临时表) DDL
CREATE TEMPORARY TABLE UserAttrSource ( 
    `data` string,
    `kafkaMetaTimestamp` TIMESTAMP(3) METADATA FROM 'timestamp', -- kafka record携带的源数据时间戳,参考官网kafka connector
    proctime 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
);

-- 创建SINKCREATE 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 AS
  SELECT 
    data,
    kafkaMetaTimestamp FROM (
      SELECT 
        *,
        ROW_NUMBER() OVER (PARTITION BY json_value(data,'$.data_hash') ORDER BY proctime DESC) as row_num
      FROM 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;

到了这里,关于[实战-10]FlinkSql 如何实现数据去重?的文章就介绍完了。如果您还想了解更多内容,请在右上角搜索TOY模板网以前的文章或继续浏览下面的相关文章,希望大家以后多多支持TOY模板网!

本文来自互联网用户投稿,该文观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如若转载,请注明出处: 如若内容造成侵权/违法违规/事实不符,请点击违法举报进行投诉反馈,一经查实,立即删除!

领支付宝红包 赞助服务器费用

相关文章

  • 2.3 如何使用FlinkSQL读取&写入到JDBC(MySQL)

    FlinkSQL允许使用 JDBC连接器,向任意类型的关系型数据库读取或者写入数据 添加Maven依赖 注意:如果使用 sql-client客户端,需保证 flink-1.17.1/lib 目录下 存在相应的jar包  相关jar可以通过官网下载:JDBC SQL 连接器  FlinkSQL读取MySQL表时,为批式处理,在流式计算任务中,通常被

    2024年02月06日
    浏览(54)
  • 数据湖Iceberg-FlinkSQL集成(5)

    数据湖Iceberg-简介(1) 数据湖Iceberg-存储结构(2) 数据湖Iceberg-Hive集成Iceberg(3) 数据湖Iceberg-SparkSQL集成(4) 数据湖Iceberg-FlinkSQL集成(5) 数据湖Iceberg-FlinkSQL-kafka类型表数据无法成功写入(6) 数据湖Iceberg-Flink DataFrame集成(7) Flink与Iceberg的版本对应关系如下 Flink 版本 Iceberg 版本 1.11 0.9.0 –

    2023年04月26日
    浏览(36)
  • 数据湖Iceberg介绍和使用(集成Hive、SparkSQL、FlinkSQL)

    概述 为了解决数据存储和计算引擎之间的适配的问题,Netflix开发了Iceberg,2018年11月16日进入Apache孵化器,2020 年5月19日从孵化器毕业,成为Apache的顶级项目。 Iceberg是一个面向海量数据分析场景的 开放表格式(Table Format) 。表格式(Table Format)可以理解为 元数据以及数据文

    2024年02月10日
    浏览(42)
  • 2.2 如何使用FlinkSQL读取&写入到文件系统(HDFS\Local\Hive)

    目录 1、文件系统 SQL 连接器 2、如何指定文件系统类型 3、如何指定文件格式 4、读取文件系统 4.1 开启 目录监控  4.2 可用的 Metadata 5、写出文件系统 5.1 创建分区表 5.2 滚动策略、文件合并、分区提交 5.3 指定 Sink Parallelism 6、示例_通过FlinkSQL读取kafka在写入hive表 6.1、创建

    2024年02月07日
    浏览(38)
  • 基于 Dinky + FlinkSQL + Flink CDC 同步 MySQL 数据到 Elasticsearch、Kafka

    Dinky 是一个开箱即用的一站式实时计算平台以 Apache Flink 为基础,连接 OLAP 和数据湖等众多框架致力于流批一体和湖仓一体的建设与实践。本文以此为FlinkSQL可视化工具。 Flink SQL 使得使用标准 SQL 开发流式应用变得简单,免去代码开发。 Flink CDC 本文使用 MySQL CDC 连接器 允许从

    2024年02月16日
    浏览(46)
  • 深入理解flinksql执行流程,calcite与catalog相关概念,扩展解析器实现语法的扩展

    flink在执行sql语句时,是无法像java/scala代码一样直接去使用的,需要解析成电脑可以执行的语言,对sql语句进行解析转化。 这里说的我感觉其实不是特别准确,应该是 flink使用的是一款开源SQL解析工具Apache Calcite ,Calcite使用Java CC对sql语句进行了解析 。 那么我们先来简单说

    2024年02月21日
    浏览(46)
  • FlinkAPI开发之FlinkSQL

    这里的依赖是一个Java的“桥接器”(bridge),主要就是负责Table API和下层DataStream API的连接支持,按照不同的语言分为Java版和Scala版。 如果我们希望在本地的集成开发环境(IDE)里运行Table API和SQL,还需要引入以下依赖: 对于Flink这样的流处理框架来说,数据流和表在结构上

    2024年01月25日
    浏览(35)
  • Flink 学习十 FlinkSQL

    flink sql 基于flink core ,使用sql 语义方便快捷的进行结构化数据处理的上层库; 类似理解sparksql 和sparkcore , hive和mapreduce 1.1 工作流程 整体架构和工作流程 数据流,绑定元数据 schema ,注册成catalog 中的表 table / view 用户使用table Api / table sql 来表达计算逻辑 table-planner利用 apache calci

    2024年02月10日
    浏览(46)
  • FlinkSQL ChangeLog

    登录sql-client,创建一个upsert-kafka的sql作业(注意,这里发送给kafka的消息必须带key,普通只有value的消息无法解析,这里的key即是主键的值) 发送消息带key和消费消息显示key方式如下 作业的DAG图如下 DAG图中有一个ChangelogNormalize,代码中搜索到对应的类是StreamPhysicalChangelogNo

    2024年03月28日
    浏览(54)
  • FLinkSQL+FlinkCDC

    流程: FlinkCDC读取MysqlBinLog,增删改同步写入Mysql表。 集群模式为Flink的Standalone模式,不安装Hadoop 一个非常好的博客 每台机器防火墙关闭 配置静态ip地址 两台机器能互相免密登录 配置hosts实现主机名映射ip地址 实现Xshell连接两台机器 配置xsync分发文件脚本 Mysql需要开启BinLog 安

    2024年02月14日
    浏览(24)

觉得文章有用就打赏一下文章作者

支付宝扫一扫打赏

博客赞助

微信扫一扫打赏

请作者喝杯咖啡吧~博客赞助

支付宝扫一扫领取红包,优惠每天领

二维码1

领取红包

二维码2

领红包