如何基于日志，同步实现数据的一致性和实时抽取?

1、背景

事情是从公司前段时间的需求提及，你们知道宜信是一家金融科技公司，咱们的不少数据与标准互联网企业不一样，大体来讲就是：

玩数据的人都知道数据是很是有价值的，而后这些数据是保存在各个系统的数据库中，如何让须要数据的使用方获得一致性、实时的数据呢？

过去的通用作法有几种，分别是：

• DBA开放各个系统的备库，在业务低峰期（好比夜间），使用方各自抽取所需数据。因为抽取时间不一样，各个数据使用方数据不一致，数据发生冲突，并且重复抽取，相信很多DBA很头疼这个事情。
• 公司统一的大数据平台，经过Sqoop 在业务低峰期到各个系通通一抽取数据， 并保存到Hive表中, 而后为其余数据使用方提供数据服务。这种作法解决了一致性问题，但时效性差，基本是T+1的时效。
• 基于trigger的方式获取增量变动，主要问题是业务方侵入性大，并且trigger也带来性能损失。

这些方案都不算完美。咱们在了解和考虑了不一样实现方式后，最后借鉴了 linkedin的思想，认为要想同时解决数据一致性和实时性，比较合理的方法应该是来自于log。

（此图来自：www.confluent.io/blog/using-…

把增量的Log做为一切系统的基础。后续的数据使用方，经过订阅kafka来消费log。

好比：

• 大数据的使用方能够将数据保存到Hive表或者Parquet文件给Hive或Spark查询；
• 提供搜索服务的使用方能够保存到Elasticsearch或HBase 中；
• 提供缓存服务的使用方能够将日志缓存到Redis或alluxio中；
• 数据同步的使用方能够将数据保存到本身的数据库中；
• 因为kafka的日志是能够重复消费的，而且缓存一段时间，各个使用方能够经过消费kafka的日志来达到既能保持与数据库的一致性，也能保证明时性；

为何使用log和kafka做为基础，而不使用Sqoop进行抽取呢？ 由于：

为何不使用dual write（双写）呢？，请参考https://www.confluent.io/blog/using-logs-to-build-a-solid-data-infrastructure-or-why-dual-writes-are-a-bad-idea/

这里就很少作解释了。

2、整体架构

因而咱们提出了构建一个基于log的公司级的平台的想法。

下面解释一下DWS平台，DWS平台是有3个子项目组成：

• Dbus（数据总线）：负责实时将数据从源端实时抽出，并转换为约定的自带schema的json格式数据(UMS 数据)，放入kafka中；
• Wormhole（数据交换平台）：负责从kafka读出数据 将数据写入到目标中；
• Swifts（实时计算平台）：负责从kafka中读出数据，实时计算，并将数据写回kafka中。

图中：

• Log extractor和dbus共同完成数据抽取和数据转换，抽取包括全量和增量抽取。
• Wormhole能够将全部日志数据保存到HDFS中； 还能够将数据落地到全部支持jdbc的数据库，落地到HBash，Elasticsearch，Cassandra等；
• Swifts支持以配置和SQL的方式实现对进行流式计算，包括支持流式join，look up，filter，window aggregation等功能；
• Dbus web是dbus的配置管理端，rider除了配置管理之外，还包括对Wormhole和Swifts运行时管理，数据质量校验等。

因为时间关系，我今天主要介绍DWS中的Dbus和Wormhole，在须要的时候附带介绍一下Swifts。

3、dbus解决方案

3.1 日志解析

如前面所说，Dbus主要解决的是将日志从源端实时的抽出。 这里咱们以MySQL为例子，简单说明如何实现。

咱们知道，虽然MySQL InnoDB有本身的log，MySQL主备同步是经过binlog来实现的。以下图：

图片来自：github.com/alibaba/can…

而binlog有三种模式：

• Row 模式：日志中会记录成每一行数据被修改的形式，而后在slave端再对相同的数据进行修改。
• Statement 模式: 每一条会修改数据的sql都会记录到 master的bin-log中。slave在复制的时候SQL进程会解析成和原来master端执行过的相同的SQL来再次执行。
• Mixed模式： MySQL会根据执行的每一条具体的sql语句来区分对待记录的日志形式，也就是在Statement和Row之间选择一种。

他们各自的优缺点以下：

此处来自：www.jquerycn.cn/a_13625

因为statement 模式的缺点，在与咱们的DBA沟经过程中了解到，实际生产过程当中都使用row 模式进行复制。这使得读取全量日志成为可能。

一般咱们的MySQL布局是采用 2个master主库（vip）+ 1个slave从库 + 1个backup容灾库 的解决方案，因为容灾库一般是用于异地容灾，实时性不高也不便于部署。

为了最小化对源端产生影响，显然咱们读取binlog日志应该从slave从库读取。

读取binlog的方案比较多，github上很多，参考https://github.com/search?utf8=%E2%9C%93&q=binlog。最终咱们选用了阿里的canal作位日志抽取方。

Canal最先被用于阿里中美机房同步， canal原理相对比较简单：

• Canal模拟MySQL Slave的交互协议，假装本身为MySQL Slave，向MySQL Slave发送dump协议
• MySQL master收到dump请求，开始推送binary log给Slave(也就是canal)
• Canal解析binary log对象(原始为byte流)

图片来自： github.com/alibaba/can…

3.2 解决方案

Dbus 的MySQL版主要解决方案以下：

对于增量的log，经过订阅Canal Server的方式，咱们获得了MySQL的增量日志：

• 按照Canal的输出，日志是protobuf格式，开发增量Storm程序，将数据实时转换为咱们定义的UMS格式(json格式,稍后我会介绍），并保存到kafka中；
• 增量Storm程序还负责捕获schema变化，以控制版本号；
• 增量Storm的配置信息保存在Zookeeper中，以知足高可用需求。
• Kafka既做为输出结果也做为处理过程当中的缓冲器和消息解构区。

在考虑使用Storm做为解决方案的时候，咱们主要是认为Storm有如下优势：

• 技术相对成熟，比较稳定，与kafka搭配也算标准组合；
• 实时性比较高，可以知足实时性需求；
• 知足高可用需求；
• 经过配置Storm并发度，能够活动性能扩展的能力；

3.3 全量抽取

对于流水表，有增量部分就够了，可是许多表须要知道最初（已存在）的信息。这时候咱们须要initial load（第一次加载）。

对于initial load（第一次加载），一样开发了全量抽取Storm程序经过jdbc链接的方式，从源端数据库的备库进行拉取。initial load是拉所有数据，因此咱们推荐在业务低峰期进行。好在只作一次，不须要天天都作。

全量抽取，咱们借鉴了Sqoop的思想。将全量抽取Storm分为了2 个部分：

• 数据分片
• 实际抽取

数据分片须要考虑分片列，按照配置和自动选择列将数据按照范围来分片，并将分片信息保存到kafka中。

下面是具体的分片策略：

全量抽取的Storm程序是读取kafka的分片信息，采用多个并发度并行链接数据库备库进行拉取。由于抽取的时间可能很长。抽取过程当中将实时状态写到Zookeeper中，便于心跳程序监控。

3.4 统一消息格式

不管是增量仍是全量，最终输出到kafka中的消息都是咱们约定的一个统一消息格式,称为UMS(unified message schema)格式。

以下图所示：

消息中schema部分，定义了namespace 是由 类型+数据源名+schema名+表名+版本号+分库号+分表号 可以描述整个公司的全部表，经过一个namespace就能惟必定位。

• _ums_op_ 代表数据的类型是I（insert），U（update），D（删除）；
• _ums_ts_ 发生增删改的事件的时间戳，显然新的数据发生的时间戳更新；
• _ums_id_ 消息的惟一id，保证消息是惟一的，但这里咱们保证了消息的前后顺序（稍后解释）；

payload是指具体的数据，一个json包里面能够包含1条至多条数据，提升数据的有效载荷。

UMS中支持的数据类型，参考了Hive类型并进行简化，基本上包含了全部数据类型。

3.5 全量和增量的一致性

在整个数据传输中，为了尽可能的保证日志消息的顺序性，kafka咱们使用的是1个partition的方式。在通常状况下，基本上是顺序的和惟一的。

可是咱们知道写kafka会失败，有可能重写，Storm也用重作机制，所以，咱们并不严格保证exactly once和彻底的顺序性，但保证的是at least once。

所以_ums_id_变得尤其重要。

对于全量抽取，_ums_id_是惟一的，从zk中每一个并发度分别取不一样的id片区，保证了惟一性和性能，填写负数，不会与增量数据冲突，也保证他们是早于增量消息的。

对于增量抽取，咱们使用的是MySQL的日志文件号 + 日志偏移量做为惟一id。Id做为64位的long整数，高7位用于日志文件号，低12位做为日志偏移量。

例如：000103000012345678。 103 是日志文件号，12345678 是日志偏移量。

这样，从日志层面保证了物理惟一性（即使重作也这个id号也不变），同时也保证了顺序性（还能定位日志）。经过比较_ums_id_ 消费日志就能经过比较_ums_id_知道哪条消息更新。

其实_ums_ts_与_ums_id_意图是相似的，只不过有时候_ums_ts_可能会重复,即在1毫秒中发生了多个操做，这样就得靠比较_ums_id_了。

3.6 心跳监控和预警

整个系统涉及到数据库的主备同步，Canal Server，多个并发度Storm进程等各个环节。

所以对流程的监控和预警就尤其重要。

经过心跳模块，例如每分钟（可配置）对每一个被抽取的表插入一条心态数据并保存发送时间，这个心跳表也被抽取，跟随着整个流程下来，与被同步表在实际上走相同的逻辑（由于多个并发的的Storm可能有不一样的分支），当收到心跳包的时候，即使没有任何增删改的数据，也能证实整条链路是通的。

Storm程序和心跳程序将数据发送公共的统计topic，再由统计程序保存到influxdb中，使用grafana进行展现，就能够看到以下效果：

图中是某业务系统的实时监控信息。上面是实时流量状况，下面是实时延时状况。能够看到，实时性仍是很不错的，基本上1~2秒数据就已经到末端kafka中。

Granfana提供的是一种实时监控能力。

若是出现延时，则是经过dbus的心跳模块发送邮件报警或短信报警。

3.7 实时脱敏

考虑到数据安全性，对于有脱敏需求的场景，Dbus的全量storm和增量storm程序也完成了实时脱敏的功能。脱敏方式有3种：

总结一下：简单的说，Dbus就是将各类源的数据，实时的导出，并以UMS的方式提供订阅， 支持实时脱敏，实际监控和报警。

4、Wormhole解决方案

说完Dbus，该说一下Wormhole，为何两个项目不是一个，而要经过kafka来对接呢？

其中很大一个缘由就是解耦，kafka具备自然的解耦能力，程序直接能够经过kafka作异步的消息传递。Dbus和Wornhole内部也使用了kafka作消息传递和解耦。

另一个缘由就是，UMS是自描述的，经过订阅kafka，任何有能力的使用方来直接消费UMS来使用。

虽然UMS的结果能够直接订阅，但还须要开发的工做。Wormhole解决的是：提供一键式的配置，将kafka中的数据落地到各类系统中，让没有开发能力的数据使用方经过wormhole来实现使用数据。

如图所示，Wormhole 能够将kafka中的UMS 落地到各类系统，目前用的最多的HDFS，JDBC的数据库和HBase。

在技术栈上， wormhole选择使用spark streaming来进行。

在Wormhole中，一条flow是指从一个namaspace从源端到目标端。一个spark streaming服务于多条flow。

选用Spark的理由是很充分的：

• Spark自然的支持各类异构存储系统；
• 虽然Spark Stream比Storm延时稍差，但Spark有着更好的吞吐量和更好的计算性能；
• Spark在支持并行计算方面有更强的灵活性；
• Spark提供了一个技术栈内解决Sparking Job，Spark Streaming，Spark SQL的统一功能，便于后期开发；

这里补充说一下Swifts的做用：

• Swifts的本质是读取kafka中的UMS数据，进行实时计算，将结果写入到kafka的另一个topic。
• 实时计算能够是不少种方式：好比过滤filter，projection（投影），lookup， 流式join window aggregation，能够完成各类具备业务价值的流式实时计算。

Wormhole和Swifts对好比下：

4.1 落HDFS

经过Wormhole Wpark Streaming程序消费kafka的UMS，首先UMS log能够被保存到HDFS上。

kafka通常只保存若干天的信息，不会保存所有信息，而HDFS中能够保存全部的历史增删改的信息。这就使得不少事情变为可能：

• 经过重放HDFS中的日志，咱们可以还原任意时间的历史快照。
• 能够作拉链表，还原每一条记录的历史信息，便于分析；
• 当程序出现错误是，能够经过回灌（backfill），从新消费消息，从新造成新的快照。

能够说HDFS中的日志是不少的事情基础。

介于Spark原生对parquet支持的很好，Spark SQL可以对Parquet提供很好的查询。UMS落地到HDFS上是保存到Parquet文件中的。Parquet的内容是全部log的增删改信息以及_ums_id_，_ums_ts_都存下来。

Wormhole spark streaming根据namespace 将数据分布存储到不一样的目录中，即不一样的表和版本放在不一样目录中。

因为每次写的Parquet都是小文件，你们知道HDFS对于小文件性能并很差，所以另外还有一个job，天天定时将这些的Parquet文件进行合并成大文件。

每一个Parquet文件目录都带有文件数据的起始时间和结束时间。这样在回灌数据时，能够根据选取的时间范围来决定须要读取哪些Parquet文件，没必要读取所有数据。

4.2 插入或更新数据的幂等性

经常咱们遇到的需求是，将数据通过加工落地到数据库或HBase中。那么这里涉及到的一个问题就是，什么样的数据能够被更新到数据？

这里最重要的一个原则就是数据的幂等性。

不管是遇到增删改任何的数据，咱们面临的问题都是：

• 该更新哪一行；
• 更新的策略是什么。

对于第一个问题，其实就须要定位数据要找一个惟一的键，常见的有：

• 使用业务库的主键；
• 由业务方指定几个列作联合惟一索引；

对于第二个问题，就涉及到_ums_id_了，由于咱们已经保证了_ums_id_大的值更新，所以在找到对应数据行后，根据这个原则来进行替换更新。

之因此要软删除和加入_is_active_列，是为了这样一种状况：

若是已经插入的_ums_id_比较大，是删除的数据（代表这个数据已经删除了）， 若是不是软删除，此时插入一个_ums_id_小的数据（旧数据），就会真的插入进去。

这就致使旧数据被插入了。不幂等了。因此被删除的数据依然保留（软删除）是有价值的，它能被用于保证数据的幂等性。

4.3 HBase的保存

插入数据到Hbase中，至关要简单一些。不一样的是HBase能够保留多个版本的数据（固然也能够只保留一个版本）默认是保留3个版本；

所以插入数据到HBase，须要解决的问题是：

• 选择合适的rowkey：Rowkey的设计是能够选的，用户能够选择源表的主键，也能够选择若干列作联合主键。
• 选择合适的version：使用_ums_id_+ 较大的偏移量（好比100亿） 做为row的version。

Version的选择颇有意思，利用_ums_id_的惟一性和自增性，与version自身的比较关系一致：即version较大等价于_ums_id_较大，对应的版本较新。

从提升性能的角度，咱们能够将整个Spark Streaming的Dataset集合直接插入到HBase，不须要比较。让HBase基于version自动替咱们判断哪些数据能够保留，哪些数据不须要保留。

Jdbc的插入数据：插入数据到数据库中，保证幂等的原理虽然简单，要想提升性能在实现上就变得复杂不少，总不能一条一条的比较而后在插入或更新。

咱们知道Spark的RDD/dataset都是以集合的方式来操做以提升性能，一样的咱们须要以集合操做的方式实现幂等性。

具体思路是：

• 首先根据集合中的主键到目标数据库中查询，获得一个已有数据集合；
• 与dataset中的集合比较，分出两类：

A：不存在的数据，即这部分数据insert就能够；

B：存在的数据，比较_ums_id_， 最终只将哪些_ums_id_更新较大row到目标数据库，小的直接抛弃。

使用Spark的同窗都知道，RDD/dataset都是能够partition的，可使用多个worker并进行操做以提升效率。

在考虑并发状况下，插入和更新均可能出现失败，那么还有考虑失败后的策略。

好比：由于别的worker已经插入，那么由于惟一性约束插入失败，那么须要改成更新，还要比较_ums_id_看是否可以更新。

对于没法插入其余状况（好比目标系统有问题），Wormhole还有重试机制。插入到其余存储中的就很少介绍了，总的原则是：根据各自存储自身特性，设计基于集合的，并发的插入数据实现。这些都是Wormhole为了性能而作的努力，使用Ｗormhole的用户没必要关心 。

5、运用案例

5.1 实时营销

说了那么多，DWS有什么实际运用呢？下面我来介绍某系统使用DWS实现了的实时营销。

如上图所示：

系统A的数据都保存到本身的数据库中，咱们知道，宜信提供不少金融服务，其中包括借款，而借款过程当中很重要的就是信用审核。

借款人须要提供证实具备信用价值的信息，好比央行征信报告，是具备最强信用数据的数据。 而银行流水，网购流水也是具备较强的信用属性的数据。

借款人经过Web或手机APP在系统A中填写信用信息时，可能会某些缘由没法继续，虽然可能这个借款人是一个优质潜在客户，但之前因为没法或好久才能知道这个信息，因此实际上这样的客户是流失了。

应用了DWS之后，借款人已经填写的信息已经记录到数据库中，并经过DWS实时的进行抽取、计算和落地到目标库中。根据对客户的打分，评价出优质客户。而后马上将这个客户的信息输出到客服系统中。

客服人员在很短的时间（几分钟之内）就经过打电话的方式联系上这个借款人（潜客），进行客户关怀，将这个潜客转换为真正的客户。咱们知道借款是有时效性的，若是时间过久就没有价值了。

若是没有实时抽取/计算/落库的能力，那么这一切都没法实现。

5.2 实时报表系统

另一个实时报表的应用以下：

咱们数据使用方的数据来自多个系统，之前是经过T+1的方式得到报表信息，而后指导次日的运营，这样时效性不好。

经过DWS，将数据从多个系统中实时抽取，计算和落地，并提供报表展现，使得运营能够及时做出部署和调整，快速应对。

6、总结

• DWS技术上基于主流实时流式大数据技术框架，高可用大吞吐强水平扩容，低延迟高容错最终一致。
• DWS能力上支持异构多源多目标系统，支持多数据格式（结构化半结构化非结构化数据）和实时技术能力。
• DWS将三个子项目合并做为一个平台推出，使得咱们具有了实时的能力， 驱动各类实时场景应用。
• 适合场景包括：实时同步／实时计算／实时监控／实时报表／实时分析／实时洞察／实时管理／实时运营／实时决策

做者：王东

7月25日晚8点，线上直播，【AI中台——智能聊天机器人平台】，点击了解详情。

来源：宜信技术学院