数据质量管理中的数据稽核平台建设(201203)

数据稽核可以理解为数据质量管理的一部分，当然也是属于数据治理的大框架范畴。在主数据管理平台的建设和实践中，往往会构建数据质量管理模块进行相关的数据质量检查等工作。

由于数据管控和治理工作本身是各个企业信息化建设中都相当关注的内容，而且大部分的业务协同，流程断点等问题往往都和数据一致性和准确性密切相关，因此数据稽核平台的建设也逐渐从主数据管理平台剥离出来，形成一个独立的系统。

在讲数据稽核前，还是给IT系统建设中的数据稽核一个简单的定义。

在IT应用建设中的数据稽核往往是指对分散在多个业务系统中的同一类数据进行的数据检查和核对过程，以解决数据准确性和一致性问题。

数据稽核概述

数据稽核是企业进行数据质量管理的一个关键举措。即首先建设数据质量管理部门和管理结构，并设置明确的质量标准和质量管控体系，并基于质量管控体系和管控流程，对数据展开稽查，稽核，比对等各种质量管理工作。

同时对发现的数据质量问题，通过流程驱动进行数据的修改和修复，在修复后进一步触发后续的持续检查和迭代，以形成一个完整的数据质量管控闭环流程。

而数据数据稽核系统作为数据质量管控中重要的环节提供技术和数据平台支撑。

数据稽核是数据质量管控的一个核心内容，重点就是实现数据的完整性和一致性检查，提升数据质量，数据稽核是一个从数据采集，预处理，比对，分析，预警，通知，问题修复的完整数据质量管控链条。

我在前面也谈到过，在当前的应用和架构下，企业业务系统间的数据集成模式导致了核心的主数据和跨系统共享的动态数据全部落地，由于本身数据集成的问题或者由于数据源头管理不善等原因导致了大量的数据不一致性。虽然一直在做数据清理工作，但是这种不一致性和问题将持续存在于整个应用架构体系里面。

数据稽核的核心过程说明

数据稽核平台核心实现逻辑并不复杂，简单来说就是采集各个业务系统的数据进行分析和比对，得出最终的稽核结果和报表，然后触发问题修改和修正，持续迭代和闭环。

因此数据集合核心过程可描述如下：

数据稽核的整个流程首先是 数据的采集和适配。

这个常见方式是通过ETL工具来完成，ETL工具采集到的数据做初步的数据清理和预处理。

在这个步骤完成后根据预定义的数据稽核和校验规则，对数据进行差异分析和异常分析，对于分析的结果， 一方面是实时的预警和通知，一方面是根据预先定义的报表模版生产数据稽核统计报表 。以上完全可以配置为一个自动化的流程，当然对于核心的业务对象或实体，我们还可以定义稽核的时间范围，稽核的业务规则进行实时的数据比对工作。

其次是 跨系统的数据比对

数据比对本身是一个由粗到细的过程，首先是数据表级别的比较，但是这个往往并不需要；然后是数据表中记录层级的数据比较，A系统同步了一条数据到B系统，是否正常成功同步到，首先要比对的就是两个数据表的key值关联是否存在。

行记录级别比较完成后是字段级别的数据比对工作，字段级的比对分为两个层面，一个是数据表表结构和字段结构元数据的一致性，如相同的表两边字段数量不一致，相同的字段的字段类型或长度不一致等；其次是字段内数据和内容的一致性比对。还有些数据稽核工具会提供数据参考完整性和通用性业务规则校验的功能，但是这个不是数据稽核的重点，更多还是应该是业务系统自身去做好参照完整性控制工作。

最后谈下数据稽核应该是一个高度可灵活配置的产品平台，其中包括了稽核流程可以配置，ETL和元数据定义，字段映射可配置；数据稽核规则可配置，报表模版可以预定义和配置；预警和通知规则和配置。有了这些灵活的可配置能力后，数据稽核平台基本就可以应用到很多类似数据稽核和比对的场景中。

大数据相关技术在数据稽核中应用

接着谈下大数据或海量数据处理技术在数据稽核整个解决方案中可能的应用场景。

对于数据采集和ETL部分

这部分可以基于传统的ETL工具或模式来完成，对于非结构化文件或日志的采集可以通过Flume等分布式开源数据采集平台来完成。对于数据采集部分主要是数据采集和传输效率的提升问题。

对于数据采集，可以搭建分布式的数据采集集群，如果对于单个ETL不拆分，那么可以对多个数据库，多个数据表的数据采集均衡的分配到多台数据采集和处理服务器上，实现数据的并行采集处理，加快数据的采集速度和降低IO瓶颈。

对于实时采集方面，可以考虑直接采用类似BinLog日志采集模式，可以实时的采集到变更数据。对于ETL而言，首先对于数据的unLoad和Load过程，最好的方法就是采用适配数据库的原始批量数据文件导出和导入接口，例如Oracle的SqlLoader，Sybase和SqlServer数据库的BCP工具等，这种原生接口对于大数据批量导出性能最好，而且导出的文件还可以在ETL处理过程中进行压缩传输。

对于上亿条记录的大表，那么单个表的导出本身也是一个相当耗时的工作。在这里我们可以考虑对这种大型表进行行拆分和列拆分操作，将一个任务作业拆分为多个子任务，每个子任务在分配到集群中的多个节点机去运行，最好进行数据的汇总处理。在这块暂时没有做过具体的试验，无法预估实际的效果和性能提升情况。

对于ETL中的Transfer部分，在Oracle的ODI工具中我们看到，已经将ETL更多的转换为了ELT模式，即在数据采集完成后在目标端数据库再做具体的数据转换和处理等相关工作。在这种情况下将比传统的ETL方法获取到更高的性能。如果仍然是采用类似ELT模式的数据采集和处理，可以考虑将转换规则进行分步骤的拆解，将步骤分解到多台节点机器进行处理再进行最终结果的归并，类似MapReduce的并行计算模式。

总之，对于数据采集部分我们期望达到的就是并行采集，并行写入，同时在处理环节进行数据分流和数据转换。在传输过程中做好数据的压缩，提升数据采集和写入的性能和速度。

对于数据处理和分析部分

对于数据分析部分是另一个很重要的内容，在大数据总体解决方案中也经常谈到数据分析部分的性能问题。在这里有商用的GreenPulm，开源的Hive，也有针对MySQL数据库的开源Infobright等。在这里的几个重点是分布式数据库集群，内存数据库，列式数据库，MPP大规模并行处理，数据库DaaS层，数据查询任务分解和汇聚等关键词。

对于数据分析部分最关心的还是在海量数据下面的查询效率问题。

对于数据稽核中存在基于某种规则的实时性的数据分析和一致性比对工作，而这种是典型的海量数据下实时查询和汇总统计过程。因此在我们对目标数据库的考虑上不再是简单的单数据库模式，而应该是一种支撑高性能和线性扩展的数据库集群模式，在这种模式下可以很好的解决数据查询的性能问题。

对于定时处理的数据稽核任务，也存在数据处理的效率问题，对于电信行业话单和计费结算数据的比对分析往往需要7，8个小时才能够完成。说明在数据处理过程中还有很多具体的性能提升点。包括数据稽核规则本身的分解，将稽核任务分配到多个节点去运行然后再进行数据的聚合。

个人认为在数据稽核中的数据处理部分MapReduce有很多具体的用武之地，而我们更多的是需要考虑如何对MapReduce框架根据典型的数据稽核规则场景进行更好的二次封装，能够方便规则更加灵活的配置和调度。

对于实时数据分析和比对

对于这个有明确的需求场景，即实时采集数据并动态的监控数据差异化和比对情况。这个将是后续数据稽核平台的一个重要发展点。在这种场景下基本基于一种持续的不落地的数据流处理方式。

实时流处理打破了传统的数据分析和处理的模式，即数据最终积累和落地后再针对海量数据进行拆分处理，然后进行分析统计，传统的模式很难真正达到实时性和速 度要求。而实时流处理模型的重点正是既有类似Hadoop中MapReduce和PIG一样的数据处理和调度引擎，由解决了直接通过输入适配接驳到流的入 口，流实时达到实时处理，实时进行分组汇聚等增量操作。

在这种模式下一个重点就是前面谈到过的对于数据稽核规则需要进行拆分，形成多个可以并行处理的PE任务，对实时达到的数据流进行处理，形成某种结果信息后再向后续节点推送，最终实时的监控和查询数据比对结果。这是一种实时动态监控的模式，对于在实时性要求高的数据质量监控中可以使用。

数据稽核平台构建

对于数据稽核出现的场景，最主要的可以理解为两个方面：

其一是同样一个基础数据或共享数据在多个业务系统中各自维护，出现数据不一致或者说无法映射的问题；其二是在我们数据集成的实现过程中，将同样一份数据分发和同步到多个业务系统中，同时由于数据集成本身出现问题，或者说目标业务系统本身允许了对数据的CUD操作都会导致随着时间的变化同样的数据变来不一致。

对于数据不一致或数据缺失带来的最大问题就是影响到跨多个IT系统的也流转。类似的场景有很多，类似项目信息从项目管理系统同步到采购系统后，可能发现项目类型本身不对导致无法根据项目下达采购订单。

或者说一个完整的采购订单从采购系统同步到ERP的时候会发现因为ERP缺失了某个物料而同步失败。这些都直接影响到我们的业务协同，而我们往往经常的应对就是出了问题才通过IT系统手工从后台修复数据或者进行数据清理后重新导入或同步。

对于做IT系统日常运维的人员来说，运维中本身硬件环境或服务器宕机的时候往往很少，而更多的时间都在手工后台修改数据，同步数据，解决数据不一致的问题。

前面刚好谈到过我们经常的思维都是解决问题应该从源头抓起，类似数据不一致的问题应该严格的控制住数据源头并制定相关的数据管理和治理规范，但是实际的情况确是从根源上解决问题的方法看起来很完美，但是短期有诸多业务本身方面的原因而无法落地。

那么在彻底从根源解决问题和出了问题再手工解决之间就有一个更好的解决方案，即我们可以提前的发现数据不一致的问题并预警，而不是真正到了业务协同出现问题再去临时应急。而一个优秀的数据稽核平台就是基于以上背景条件产生的。

一个最小化的数据稽核系统核心应该包括三个方面的内容： 其一是数据采集和存储，其二是规则引擎，其三是数据分析和展示。 而一个高效的数据稽核系统必须具备数据不一致的准实时预警能力，对于准实时的实现本身又包括了数据采集的准实时和数据分析的高效性能。

首先来谈下数据采集和存储 ，前面谈到过基于ETL技术来实现数据采集，在ETL方式下要实现数据本身的准实时采集是相当困难的。而真正可以考虑的方式则是类似数据实时复制和同步技术，对于Oracle数据库我们可以采用Logminer实时捕获增量归档日志信息，对于Mysql数据库可以采用Binlog日志分析等。但是这里面一个重点就是需要对业务系统本身的DB数据库开放相关的权限和初始化参数，能否推动业务系统去做这个调整本身存在困难。

采集回来的数据存储在哪里？

常规做法是仍然存储在关系型数据库里面，那么我们可以想象下如果对于两张相关结构的有50个字段的数据库表。两张表如果都有1000万条数据，如果我们要比较出两张表存在的数据差异，对于这种大量的全部扫描和比对，性能本身是存在大问题的。

正是由于这个原因，我们可以考虑将采集来的数据直接导入到HDFS库里面，然后采用大数据分析方法和技术进行数据比对。而数据入到HDFS，当前的Sqoop组件是支持增量数据采集的，即可以通过我们制定的关键字段对源表数据进行增量采集。这种采集方式虽然比不上实时的数据同步复制，但是满足准实时性的要求基本不会有太大的问题。

其次，我们来谈下数据分析 ，对于这种数据稽核本身的数据量如果在大数据环境里面本身就是小数据了，你当然可以采用Hive进行数据比对分析，但是性能估计也很难好太多。更好的方法则是采用Spark进行相应的数据比对分析，由于Spark本身是基于内存计算的，对于这种量级在内存中进行不对分析是最合适不过的。同时当前Spark本身就已经支持Spark Sql和Hive集成，也支持直接的JDBC集成等。这些都为数据比对分析上提供了相当的方便性。

还有一种设想就是是否可以将数据库的变更日志信息直接送到类似kafka的消息中间件中，采用通过spark和kafka的集成即可以实现类似数据采集的流处理和后续的内存计算分析，整个过程中由于数据本身不会二次落地都直接在内存中完成，整体性能的提升应该更大。

最后，谈下数据稽核平台里面另外一个重要的内容即规则引擎 ，规则引擎本身就是需要将数据稽核的各种校验方式和规则全部做到可配置化，即任何一个稽核流程都可以通过规则配置出来。比如我们需要检查三个业务系统中三张表数据的一致性，具体的检验规则是如何的，转换规则是如何的，当发现不一致的时候预警规则是如何的都需要做到灵活可配置。

可配置的数据稽核规则最终需要动态来生成数据稽核算法，或者说动态来生成稽核SQL语句等，这个也是整个稽核平台里面最难的部分，只有实现了该点整个数据稽核系统才谈得上是一个可以灵活可扩展的稽核平台。

对于数据稽核的结果，稽核平台本身应该提供上层分析报表能力，但是如果考虑到稽核结果的实时性，即数据稽核的结果应该可以通过消息发布订阅的方式向外部发布。只要订阅了某类数据稽核结果的订阅方，都可以实时的收集到最终的数据稽核结果信息。

将Spark和流处理，消息中间件，规则引擎等技术应用到数据稽核系统绝对不是大材小用，而是大数据技术能够更好的服务于传统业务系统，解决业务系统问题的重要尝试。即并不是我们的业务系统必须到了海量数据存储后才谈得上用大数据技术，而是在业务需求本身在实时性，效率，架构可扩展性等多个方面都可以采用大数据技术来解决真实业务场景下的业务问题。