共计 9423 个字符,预计需要花费 24 分钟才能阅读完成。
作者:王东
宜信技术研发中心架构师
- 目前就职于宜信技术研发中心,任架构师,负责流式计算和大数据业务产品解决方案。
- 曾任职于 Naver china(韩国最大搜索引擎公司)中国研发中心资深工程师,多年从事 CUBRID 分布式数据库集群开发和 CUBRID 数据库引擎开发 http://www.cubrid.org/blog/news/cubrid-cluster-introduction/
主题简介:
- DWS 的背景介绍
- dbus+wormhole 总体架构和技术实现方案
- DWS 的实际运用案例
大家好,我是王东,来自宜信技术研发中心,这是我来社群的第一次分享,如果有什么不足,请大家多多指正、包涵。
本次分享的主题是《基于日志的 DWS 平台实现和应用》,主要是分享一下目前我们在宜信做的一些事情。这个主题里面包含到 2 个团队很多兄弟姐妹的努力的结果(我们团队和山巍团队的成果)。这次就由我代为执笔,尽我努力给大家介绍一下。
其实整个实现从原理上来说是比较简单的,当然也涉及到不少技术。我会尝试用尽量简单的方式来表达,让大家了解这个事情的原理和意义。在过程中,大家有问题可以随时提出,我会尽力去解答。
DWS 是一个简称,是由 3 个子项目组成,我稍后做解释。
事情是从公司前段时间的需求说起,大家知道宜信是一个互联网金融企业,我们的很多数据与标准互联网企业不同,大致来说就是:
玩数据的人都知道数据是非常有价值的,然后这些数据是保存在各个系统的数据库中,如何让需要数据的使用方得到一致性、实时的数据呢?
- DBA 开放各个系统的备库,在业务低峰期(比如夜间),使用方各自抽取所需数据。由于抽取时间不同,各个数据使用方数据不一致,数据发生冲突,而且重复抽取,相信不少 DBA 很头疼这个事情。
- 公司统一的大数据平台,通过 Sqoop 在业务低峰期到各个系统统一抽取数据,并保存到 Hive 表中, 然后为其他数据使用方提供数据服务。这种做法解决了一致性问题,但时效性差,基本是 T + 1 的时效。
- 基于 trigger 的方式获取增量变更,主要问题是业务方侵入性大,而且 trigger 也带来性能损失。
这些方案都不算完美。我们在了解和考虑了不同实现方式后,最后借鉴了 linkedin 的思想,认为要想同时解决数据一致性和实时性,比较合理的方法应该是来自于 log。
(此图来自:https://www.confluent.io/blog/using-logs-to-build-a-solid-data-infrastructure-or-why-dual-writes-are-a-bad-idea/)
把增量的 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/
我这里就不多做解释了。
于是我们提出了构建一个基于 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。
如前面所说,Dbus 主要解决的是将日志从源端实时的抽出。这里我们以 MySQL 为例子,简单说明如何实现。
我们知道,虽然 MySQL InnoDB 有自己的 log,MySQL 主备同步是通过 binlog 来实现的。如下图:
图片来自:https://github.com/alibaba/canal
而 binlog 有三种模式:
- Row 模式:日志中会记录成每一行数据被修改的形式,然后在 slave 端再对相同的数据进行修改。
- Statement 模式: 每一条会修改数据的 sql 都会记录到 master 的 bin-log 中。slave 在复制的时候 SQL 进程会解析成和原来 master 端执行过的相同的 SQL 来再次执行。
- Mixed 模式:MySQL 会根据执行的每一条具体的 sql 语句来区分对待记录的日志形式,也就是在 Statement 和 Row 之间选择一种。
他们各自的优缺点如下:
此处来自:http://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 流)
图片来自:https://github.com/alibaba/canal
Dbus 的 MySQL 版主要解决方案如下:
对于增量的 log,通过订阅 Canal Server 的方式,我们得到了 MySQL 的增量日志:
- 按照 Canal 的输出,日志是 protobuf 格式,开发增量 Storm 程序,将数据实时转换为我们定义的 UMS 格式 (json 格式, 稍后我会介绍),并保存到 kafka 中;
- 增量 Storm 程序还负责捕获 schema 变化,以控制版本号;
- 增量 Storm 的配置信息保存在 Zookeeper 中,以满足高可用需求。
- Kafka 既作为输出结果也作为处理过程中的缓冲器和消息解构区。
在考虑使用 Storm 作为解决方案的时候,我们主要是认为 Storm 有以下优点:
- 技术相对成熟,比较稳定,与 kafka 搭配也算标准组合;
- 实时性比较高,能够满足实时性需求;
- 满足高可用需求;
- 通过配置 Storm 并发度,可以活动性能扩展的能力;
对于流水表,有增量部分就够了,但是许多表需要知道最初(已存在)的信息。这时候我们需要 initial load(第一次加载)。
对于 initial load(第一次加载),同样开发了全量抽取 Storm 程序通过 jdbc 连接的方式,从源端数据库的备库进行拉取。initial load 是拉全部数据,所以我们推荐在业务低峰期进行。好在只做一次,不需要每天都做。
全量抽取,我们借鉴了 Sqoop 的思想。将全量抽取 Storm 分为了 2 个部分:
- 数据分片
- 实际抽取
数据分片需要考虑分片列,按照配置和自动选择列将数据按照范围来分片,并将分片信息保存到 kafka 中。
下面是具体的分片策略:
全量抽取的 Storm 程序是读取 kafka 的分片信息,采用多个并发度并行连接数据库备库进行拉取。因为抽取的时间可能很长。抽取过程中将实时状态写到 Zookeeper 中,便于心跳程序监控。
无论是增量还是全量,最终输出到 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 类型并进行简化,基本上包含了所有数据类型。
在整个数据传输中,为了尽量的保证日志消息的顺序性,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_了。
整个系统涉及到数据库的主备同步,Canal Server,多个并发度 Storm 进程等各个环节。
因此对流程的监控和预警就尤为重要。
通过心跳模块,例如每分钟(可配置)对每个被抽取的表插入一条心态数据并保存发送时间,这个心跳表也被抽取,跟随着整个流程下来,与被同步表在实际上走相同的逻辑(因为多个并发的的 Storm 可能有不同的分支),当收到心跳包的时候,即便没有任何增删改的数据,也能证明整条链路是通的。
Storm 程序和心跳程序将数据发送公共的统计 topic,再由统计程序保存到 influxdb 中,使用 grafana 进行展示,就可以看到如下效果:
图中是某业务系统的实时监控信息。上面是实时流量情况,下面是实时延时情况。可以看到,实时性还是很不错的,基本上 1~2 秒数据就已经到末端 kafka 中。
Granfana 提供的是一种实时监控能力。
如果出现延时,则是通过 dbus 的心跳模块发送邮件报警或短信报警。
考虑到数据安全性,对于有脱敏需求的场景,Dbus 的全量 storm 和增量 storm 程序也完成了实时脱敏的功能。脱敏方式有 3 种:
总结一下:简单的说,Dbus 就是将各种源的数据,实时的导出,并以 UMS 的方式提供订阅,支持实时脱敏,实际监控和报警。
说完 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 对比如下:
通过 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 文件,不必读取全部数据。
常常我们遇到的需求是,将数据经过加工落地到数据库或 HBase 中。那么这里涉及到的一个问题就是,什么样的数据可以被更新到数据?
这里最重要的一个原则就是数据的幂等性。
无论是遇到增删改任何的数据,我们面临的问题都是:
- 该更新哪一行;
- 更新的策略是什么。
对于第一个问题,其实就需要定位数据要找一个唯一的键,常见的有:
- 使用业务库的主键;
- 由业务方指定几个列做联合唯一索引;
对于第二个问题,就涉及到_ums_id_了,因为我们已经保证了_ums_id_大的值更新,因此在找到对应数据行后,根据这个原则来进行替换更新。
之所以要软删除和加入_is_active_列,是为了这样一种情况:
如果已经插入的_ums_id_比较大,是删除的数据(表明这个数据已经删除了),如果不是软删除,此时插入一个_ums_id_小的数据(旧数据),就会真的插入进去。
这就导致旧数据被插入了。不幂等了。所以被删除的数据依然保留(软删除)是有价值的,它能被用于保证数据的幂等性。
插入数据到 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 为了性能而做的努力,使用Wormhole 的用户不必关心。
说了那么多,DWS 有什么实际运用呢?下面我来介绍某系统使用 DWS 实现了的实时营销。
如上图所示:
系统 A 的数据都保存到自己的数据库中,我们知道,宜信提供很多金融服务,其中包括借款,而借款过程中很重要的就是信用审核。
借款人需要提供证明具有信用价值的信息,比如央行征信报告,是具有最强信用数据的数据。而银行流水,网购流水也是具有较强的信用属性的数据。
借款人通过 Web 或手机 APP 在系统 A 中填写信用信息时,可能会某些原因无法继续,虽然可能这个借款人是一个优质潜在客户,但以前由于无法或很久才能知道这个信息,所以实际上这样的客户是流失了。
应用了 DWS 以后,借款人已经填写的信息已经记录到数据库中,并通过 DWS 实时的进行抽取、计算和落地到目标库中。根据对客户的打分,评价出优质客户。然后立刻将这个客户的信息输出到客服系统中。
客服人员在很短的时间(几分钟以内)就通过打电话的方式联系上这个借款人(潜客),进行客户关怀,将这个潜客转换为真正的客户。我们知道借款是有时效性的,如果时间太久就没有价值了。
如果没有实时抽取 / 计算 / 落库的能力,那么这一切都无法实现。
另外一个实时报表的应用如下:
我们数据使用方的数据来自多个系统,以前是通过 T + 1 的方式获得报表信息,然后指导第二天的运营,这样时效性很差。
通过 DWS,将数据从多个系统中实时抽取,计算和落地,并提供报表展示,使得运营可以及时作出部署和调整,快速应对。
说了那么多,大致总结一下:
- DWS 技术上基于主流实时流式大数据技术框架,高可用大吞吐强水平扩容,低延迟高容错最终一致。
- DWS 能力上支持异构多源多目标系统,支持多数据格式(结构化半结构化非结构化数据)和实时技术能力。
- DWS 将三个子项目合并作为一个平台推出,使得我们具备了实时的能力,驱动各种实时场景应用。
适合场景包括: 实时同步/实时计算/实时监控/实时报表/实时分析/实时洞察/实时管理/实时运营/实时决策
感谢大家的聆听,此次分享到此为止。
Q1:Oracle log reader 有开源方案吗?
A1: 对于 Oracle 业界也有许多商业解决方案,例如:Oracle GoldenGate(原来的 goldengate), Oracle Xstream, IBM InfoSphere Change Data Capture(原来的 DataMirror),Dell SharePlex (原来的 Quest),国内的 DSG superSync 等,开源的方案好用的很少。
Q2: 这个项目投入了多少人力物力?感觉有点复杂。
Q2:DWS 是三个子项目组成,平均每个项目 5~7 人。是有点复杂,其实也是试图使用大数据技术来解决我们公司目前遇到的困难。
因为是搞大数据相关技术,所有团队里面的兄弟姐妹都还是比较 happy 的:)
其实这里面,Dbus 和 Wormhole 相对固定模式化,容易轻松复用。Swifts 实时计算是与每个业务相关比较大的,自定义比较强,相对比较麻烦一些。
Q3: 宜信的这个 DWS 系统会开源么?
A3: 我们也考虑过向社区贡献,就像宜信的其他开源项目一样,目前项目刚刚成形,还有待进一步磨炼,我相信未来的某个时候,我们会给它开源出来。
Q4: 架构师怎么理解,是不是系统工程师?
A4: 不是系统工程师,在我们宜信有多位架构师,应该算是以技术驱动业务的技术管理人员。包含产品设计,技术管理等。
Q5: 复制方案是否是 OGG?
A5:OGG 与上面提到的其他商业解决方案都是可选方案。
文章出处:DBAplus 社群(dbaplus)
![post-qrcode](https://www.xgss.net/wp-content/uploads/2023/07/二维码-微信公众号256x256.jpg)