数据一致性比对(番外)

本文涉及的产品
大数据开发治理平台DataWorks,Serverless资源组抵扣包300CU*H
简介: 我写过很多如何去对数、如何批量对数的技术文档,最近项目遇到这个问题,我才发现在官方博客上还没有发布过这个课题的文章。这就像灯下黑,太长用到的知识点,反而没有意识到其重要性。

我写过很多如何去对数、如何批量对数的技术文档,最近项目遇到这个问题,我才发现在官方博客上还没有发布过这个课题的文章这就像灯下黑,太长用到的知识点,反而没有意识到其重要性。

注:这里对数的场景就是指在阿里云平台使用dataworks等大数据开发工具集成业务系统数据库(oracle等)数据上云到maxcompute的场景,所以,示例的SQL也是针对maxcompute

先说说一般业务上怎么对数的,我们做了一个报表,出了一个数据“某个产品卖了30个”。这个不只是在大数据平台上有这个数据,在业务系统也有这个数据,这些统计动作在业务系统通过程序和人工也会有一份,一般做好报表后会先对这个数据。

所以,第一线反馈回来的数据就是这个汇总数据不一致的问题。然而这个结果是非常概括的,因为就像我感觉这个月工资少发了5毛一样,如果我不看我的工资条我其实不知道自己是不是少发了。工资条不只是一个汇总数据,里面有我税前工资、奖金(浮动)、社保、扣税等一系列的明细数据,这些数据让我去判断我是不是少了5毛,而加工过的数据是复杂的。

说到这里,我其实就像表达一个事情,对数是要对明细数据。这是一切计算后事实的基础,可以拿出来作证的。

所以,两边都查一下这个汇总值使用的表的对应的记录,比如说查询“今天这个产品ID的售卖记录”。结果就发现业务系统有31笔,而大数据平台有30笔。

即便到了这里,其实我们仍然不知道期间发生了什么,为什么会丢失数据。另外我们还不知道其他商品ID的数据是不是也有丢失的,还有其他的表的数据是不是也会发生类似的情况。

1.明细数据比对

既然最终都是对明细数据,那么我是不是可以直接比对明细数据呢?回答是:正确。

一般发生这种情况,首先要比对业务系统和大数据平台两个表的数据。

image.png

1-再利用全量集成工具,从业务系统的数据库全量抽取一遍数据到大数据平台。比对数据一定要把数据放到一起,隔空比对是不存在的。因为大数据平台的容量是数百倍于业务系统的,所以,一般都在大数据平台比对。(这里有一个悖论,如果集成工具本身就有缺陷,导致抽取过程中就丢数据,岂不是永远没办法比对了。所以,如果对这个工具也不确定就得从数据库导出数据到文件,然后再加载到某个数据库上去比对。在这里,通过我对离线集成这个产品的常年使用经验,这个工具是非常可靠的,还未遇到过这个问题。)

image.png

2-根据主键关联,比对2个表中的主键的差异。如果是上面提到的记录丢失的问题,这一步做完就很容易比对出来了。这里还会发现一个问题,就是业务系统的表是不断变化的,所以,这时与大数据平台的表对比会有差异。这个差异的核心原因是:大数据平台的表是业务系统表在每日的日末(00:00:00)的一个时点数据,而业务系统的数据是一直在变化的。所以,即便有差异超出预期也不要惊慌。如果是使用实时同步可以从归档日志中获取到这期间数据的每一条变化,可以追溯变化原因。如果没有实时同步,也可以通过表中与时间相关字段去判断数据是否被更新掉。要是什么都没有(这种情况也是存在的),那就去骂骂设计表的业务系统开发(没错,是他们的锅),也可以跟业务去详细了解一下,这行记录是不是今天做的,而不是昨天。

image.png

3-还有一种情况,就是主键一致,数据内容(主键之外的字段)不一致。这种情况,还是需要考虑数据变化的情况,可以从日志、时间字段、业务等几个角度去比对。如果发现数据确实不符合预期,就需要查询同步工具的问题。

2.比对SQL分析

在上面的章节,我描述了比对今天新抽取的全量表和上日在maxcompute上使用前日全量和上日增量合并的上日全量的环节。比对两张表集合是否一致的SQL方法其实比较简单,大家第一时间就会想到集合操作。在oracle里面有Minus、except,同样在maxcompute里面也有。但是为了便于分析问题,我还是自己写了一个SQL。示例SQL(maxcompute sql)如下:

--限定日期分区,比对上日

select count(t1.BATCH_NUMBER) as cnt_left

,count(t2.BATCH_NUMBER) as cnt_right

,count(concat(t1.BATCH_NUMBER,t2.BATCH_NUMBER)) as pk_inner

,count(case when t1.BATCH_NUMBER is not null and t2.BATCH_NUMBER is null then 1 end) as pk_left

,count(case when t2.BATCH_NUMBER is not null and t1.BATCH_NUMBER is null then 1 end) as pk_right

,count(case when nvl(t1.rec_id ,'') = nvl(t2.rec_id ,'') then 1 end) as col_diff_rec_id

,count(case when nvl(t2.rec_creator ,'') = nvl(t1.rec_creator ,'') then 1 end) as col_diff_rec_creator

,count(case when nvl(t2.rec_create_time,'') = nvl(t1.rec_create_time,'') then 1 end) as col_diff_rec_create_time

from ods_dev.o_rz_lms_im_timck01 t1 -- 开发环境重新初始化的今天数据

full join ods.o_rz_lms_im_timck01 t2 -- 生产环节昨日临时增量合并的数据

on t1.BATCH_NUMBER =t2.BATCH_NUMBER

and t1.IN_STOCK_TIME =t2.IN_STOCK_TIME

and t1.OP_NO =t2.OP_NO

and t1.STOCK_CODE =t2.STOCK_CODE

and t1.YP_ID =t2.YP_ID

and t2.ds='20230426'

where t1.ds='20230426'

;

--cnt_left 9205131 说明:左表有记录数 9205131

--cnt_right 9203971 说明:右表有记录数 9203971

--pk_inner 9203971 说明:主键关联一致记录数 9203971

--pk_left 1160 说明:左表比右表多记录数 1160

--pk_right 0 说明:右表比左表多有记录数 0

--col_diff_rec_id 9203971 说明:字段一致记录数与主键一致相同,说明关联上的两个表该字段一致

--col_diff_rec_creator 9203971 说明:同上

--col_diff_rec_create_time 9203971 说明:同上

 

在上面的例子中,左表是今天重新初始化的数据,右表是在maxcomputemerge的上日全量数据。在比对之前,我们其实就应该了解这两个表的数据必然是不一致的。虽然是同一张表,但是时点是不一致的。

不一致包括几种:

1-t1表中存在的主键,在t2表中不存在;

2-t2表中存在的主键,在t1表中不存在;

3-t1t2表中都存在的主键,但是主键之外的字段值不一致;

4-t1t2表中都存在的主键,但是主键之外的字段值一致;

 

除去第4中情况,其他3种状态都是两个表的值不一致,都需要进一步核查。如果是正常情况下,第1种情况是在今天零点以后新insert到业务库中的数据,第2种情况是今天零点以后delete的业务库中的数据,第3种情况是今天零点之后update的数据,第4种情况是业务表中今天零点之后未被更新的数据。

了解这些状况后,我们就可以依据可循去识别

3.Dataworks实时同步日志表

如果同步才有的是DataWorks的离线同步,其实观测以上数据变化是有些困难的如果我们采用实时同步,数据库数据的变化是都会保留下来的。上一章节中提到的变化,都可以从日志中观测到。下面SQL,就是我查询这个变化使用的SQL,查询(dataworks实时数据同步日志表)示例SQL如下:

select from_unixtime(cast(substr(to_char(_execute_time_),1,10) as bigint)) as yy

,get_json_object(cast(_data_columns_ as string),"$.rec_id") item0

,x.*

from ods.o_rz_lms_odps_first_log x -- 实时同步数据源 o_rz_lms 的日志表

where year='2023' and month='04' and day>='10' --数据区间限制

--and hour ='18'

and _dest_table_name_='o_rz_lms_im_timck01' --数据表限制

-- 以下为主键字段

and get_json_object(cast(_data_columns_ as string),"$.yp_id") ='L1'

and get_json_object(cast(_data_columns_ as string),"$.batch_number") ='Y1'

and get_json_object(cast(_data_columns_ as string),"$.in_stock_time") ='2'

and get_json_object(cast(_data_columns_ as string),"$.op_no") ='9'

and get_json_object(cast(_data_columns_ as string),"$.stock_code") ='R'

--and _operation_type_='D'

order by _execute_time_ desc 

limit 1000

;

image.png

-- _execute_time_ 数据操作时间

-- _operation_type_ 操作类型 增删改UDI

-- _sequence_id_ 序列号,不会重复

-- _before_image_ 修改前数据

-- _after_image_ 修改后数据

-- _dest_table_name_ 操作的表名

-- _data_columns_ 操作的数据内容JSON

DataWorks的实时同步的数据源每隔一段时间就会实时写入到一张以“数据源名+_odps_first_log命名的表中,表有年、月、日、时四级分区该表的主键并不是数据操作时间,而是序列号“_execute_time_”,所以,一行数据主键的更新顺序是按照“_execute_time_”更新。

一行数据更新是有前后两个状态的,所以有“_before_image_ 修改前数据、_after_image_ 修改后数据”这两个字段来标识前后状态。

数据是在字段“_data_columns_”中,以JSON格式存储。为了识别其中的某一行数据,我使用函数解析了对应的字段,以此来确定自己要的数据。

4.持续的质量保障

到这里,我并未讲如何去处理数据不一致的情况如果确实发现数据不一致,可用的处理方式就是重新初始化全量数据。在这里要强调一点,如果离线全量集成工具是可信的,全量初始化的数据就不会丢。但是如果这种方式不可信,则就要更换方法。

在很多情况下,源端做一些业务变更也会偶发数据异常。在数据丢失原因未查明的情况下,需要经常的去做数据一致性的比对,做到防患于未然。所以,日常监控数据数据一致性也是非常重要的。

相关实践学习
基于MaxCompute的热门话题分析
本实验围绕社交用户发布的文章做了详尽的分析,通过分析能得到用户群体年龄分布,性别分布,地理位置分布,以及热门话题的热度。
一站式大数据开发治理平台DataWorks初级课程
DataWorks 从 2009 年开始,十ー年里一直支持阿里巴巴集团内部数据中台的建设,2019 年双 11 稳定支撑每日千万级的任务调度。每天阿里巴巴内部有数万名数据和算法工程师正在使用DataWorks,承了阿里巴巴 99%的据业务构建。本课程主要介绍了阿里巴巴大数据技术发展历程与 DataWorks 几大模块的基本能力。 课程目标  通过讲师的详细讲解与实际演示,学员可以一边学习一边进行实际操作,可以深入了解DataWorks各大模块的使用方式和具体功能,让学员对DataWorks数据集成、开发、分析、运维、安全、治理等方面有深刻的了解,加深对阿里云大数据产品体系的理解与认识。 适合人群  企业数据仓库开发人员  大数据平台开发人员  数据分析师  大数据运维人员  对于大数据平台、数据中台产品感兴趣的开发者
目录
相关文章
|
6月前
|
运维 测试技术 Serverless
上一任留下的 Eureka,我该如何提升她的性能和稳定性(含数据比对)?
上一任留下的 Eureka,我该如何提升她的性能和稳定性(含数据比对)?
94743 124
|
28天前
|
消息中间件 缓存 中间件
缓存一致性问题,这么回答肯定没毛病!
缓存一致性问题,这么回答肯定没毛病!
|
3月前
|
消息中间件 Java Kafka
Kafka不重复消费的终极秘籍!解锁幂等性、偏移量、去重神器,让你的数据流稳如老狗,告别数据混乱时代!
【8月更文挑战第24天】Apache Kafka作为一款领先的分布式流处理平台,凭借其卓越的高吞吐量与低延迟特性,在大数据处理领域中占据重要地位。然而,在利用Kafka进行数据处理时,如何有效避免重复消费成为众多开发者关注的焦点。本文深入探讨了Kafka中可能出现重复消费的原因,并提出了四种实用的解决方案:利用消息偏移量手动控制消费进度;启用幂等性生产者确保消息不被重复发送;在消费者端实施去重机制;以及借助Kafka的事务支持实现精确的一次性处理。通过这些方法,开发者可根据不同的应用场景灵活选择最适合的策略,从而保障数据处理的准确性和一致性。
221 9
|
3月前
|
算法 Shell
数据魔术师:如何在ClkLog中恢复丢失数据并实现数据更新
​ 在数字化的世界里,数据就是企业的血液,是推动业务发展的关键动力。想象一下,你正在运行你的业务,依赖ClkLog为你提供的数据,突然,由于网络波动或其他原因,定时脚本未能执行,页面上的数据缺失了。或者你刚刚优化了你的算法,但你需要重新计算以前的数据以便与新的算法保持一致。这种情况下,数据的完整性和稳定性就显得尤为重要,它们不仅影响业务的正常运行,而且直接关系到业务决策的准确性和及时性。
数据魔术师:如何在ClkLog中恢复丢失数据并实现数据更新
|
3月前
|
消息中间件 Java 数据处理
揭秘Apache Flink的Exactly-Once神技:如何在数据流海中确保每条信息精准无误,不丢不重?
【8月更文挑战第26天】Apache Flink 是一款先进的流处理框架,其核心特性 Exactly-Once 语义保证了数据处理的精准无误。尤其在金融及电商等高要求场景下,该特性极为关键。本文深入解析 Flink 如何实现 Exactly-Once 语义:通过状态管理确保中间结果可靠存储;利用一致的检查点机制定期保存状态快照;以及通过精确的状态恢复避免数据重复处理或丢失。最后,提供一个 Java 示例,展示如何计算用户访问次数,并确保 Exactly-Once 语义的应用。
84 0
|
3月前
|
监控 中间件
分布式链路监控系统问题之当某个Segment数据缺失时还原调用树的问题如何解决
分布式链路监控系统问题之当某个Segment数据缺失时还原调用树的问题如何解决
|
3月前
|
监控 Java 中间件
分布式链路监控系统问题之当某个Segment数据缺失的问题如何解决
分布式链路监控系统问题之当某个Segment数据缺失的问题如何解决
|
4月前
|
存储 缓存 NoSQL
架构设计篇问题之在数据割接过程中,多线程处理会导致数据错乱和重复问题如何解决
架构设计篇问题之在数据割接过程中,多线程处理会导致数据错乱和重复问题如何解决
|
存储 SQL 搜索推荐
【其他】多维分析预汇总应该怎样做才管用?
【其他】多维分析预汇总应该怎样做才管用?
37 0
|
11月前
|
消息中间件 缓存 Java
虾皮一面:如何保证数据双写一致?
虾皮一面:如何保证数据双写一致?
85 1