RDP的全称是Real-time Data Pipeline，是一个从关系数据库MySQL实时同步数据到NoSQL系统的数据管道。正如这个名字一样，RDP不生产数据，我们只是数据的“搬运工”。谈到数据传输管道，有几个基本的特性是使用方最关注的:
• 可用性
也就是系统可以对使用方提供什么样的 SLA。在系统所承诺的 SLA 之外，使用方可以通过自己额外的设计、workaround 来达成它们自己对其它系统(或者业务指标)的 SLA 承诺。
• 实时性
作为搬运工，“送货”时效如何?系统使用方注重总体端到端的延时。系统开发者，除了端到端，我们还关心每一个环节的延时，以及各个环节间交互是否合理、是否高效?
• 追溯性
作为“管道”，爬进管子里面去调查问题，或者凭经验规律揣测，显然都是比较不友好的体验。比较理想的状况是系统通过自证和他证两个方面，来达到可以让系统使用方、系统维护者可以轻松准确地找到问题所在。
总述
本文是RDP介绍系列的第三篇，我们首先从RDP团队理解追溯性的角度，来分解RDP中关于追溯性部分的设计及关键实现。那有人可能会想，高可用和高性能是我们设计一个系统时比较关注的环节，这个“追溯性”貌似是一个生僻的说法，这里我们不是谈理想和情怀。我们认为“追溯性”是一个系统，尤其是流式数据处理系统中很重要的一环！数据丢没丢？丢了哪些？在哪个环节丢的？如何在系统长期运行的过程中自证数据没有出现问题（为自愈打基础）？上下游系统遇见问题时，我们系统如何提供他证能力，让别人轻松核实一些信息？如果你有较大规模数据处理系统的建设经验，你或许能深切体会到我说的这些痛。
如上图所示，RDP系统中，我们把追溯性分解为两个方向的含义：要让数据可追溯，每一条数据（实体）必须有一个ID，而且这个ID必须在同一个租户范围内唯一，全局递增的Sequence No.就是最简单的ID。在大多数分布式系统中，我们对数据ID的命名通常是epoch+Seq No.，RDP没有这么做是因为在“高可用”一文中我们已经保证了在进程重启和切换过程中，都能保证这个Sequence No.是单调递增的，有了ID，每一条数据都有了“身份证”。RDP中关于追溯性的第二个含义是数据要有上下文，也就是当我们处理第N条数据的时候，如果我们通过简单推理知晓了第N+1条数据的信息，那么当第N+1条数据到来的时候，我们就可以验证N+1这条数据是否是我们预期的。通俗一点讲：
RDP要保证数据可追溯，就是要“治理”数据，“治”的手段是给每一条数据加上唯一的ID，“理”的手段是每一条数据自带上下文，出没出问题从上下文中有据可循。
紧接着我们分开来讲这两点。
RDP如何给数据分配ID？
RDP内部，数据流转的环节还是比较多。如下图所示: 我们在RDP MySQL Slave模块收到Binlog事件并组合成一个完整事务单位后，就分配一个ID给这个事务。也就是说我们从数据的源头上先把这个事情做了，那么后面多个环节间的流转，我们可以全程追踪。熟悉MySQL数据库的同学可能会问，每一个事务都有GTID了，为什么还需要这个简陋的ID标识？主要原因是GTID是UUID:Sequence这样一个元组组成的，而且在某些情况下Sequence是不连续的。另外，如果MySQL发生了主从切换，那么相邻两个事务的UUID部分也是不同的。也就是说前后两个事务之间的GTID是一种弱关联，维护和运算起来也挺麻烦的，这就是我们采用简单，全局单调递增ID标识事务的原因。
事务ID作为数据的一部分，不仅会发给下游系统，也会随着checkpoint将这个checkpoint包含的最大事务ID存储到Zookeeper中。同时在RDP故障恢复过程中，也会从下游Kafka系统去比对这个事务ID，从而保证新的工作周期内事务ID还是单调递增的。
RDP如何建立事务的“上下文”？
前面已经提到，数据的“上下文”是RDP自证和他证的基础，RDP的事务上下文是建立在MySQL Binlog Event的基础之上的。如下图所示：
每一个MySQL Binlog事件的头部，包含了一个Log Position字段，其含义是下一个Binlog事件的开始Offset。而Event Size字段又代表当前Binlog事件的长度，那么很简单：
Log Position – Event Size = 当前Event的Offset
这样，通过MySQL Binlog自身附带的属性，我们就解决了在同一个Binlog文件内部“上下文”的问题。
问题说到这里，那大家自然会问，那不同Binlog文件之间的“上下文”如何建立呢？如下图所示，每一个MySQL的Binlog文件都遵循：
4字节的MagicNumber，一个Format Description Event，一个PreviousGTIDs Event…这样一个Binlog Events序列。
而Previous GTIDs Event表示的就是在这个Binlog文件之前，已经包含的所有GTID，是一个GTIDSET。那这样一来，我们也就解决了跨Binlog文件前后的可追溯性问题。RDP可以在每一个事件到来的时候，通过跨Binlog文件+Binlog文件内的前后链式属性来判断是否出现了异常！系统达到了自证的基本要求（知道自己是否健康还是已经病入膏肓了）。
到目前为止，RDP基本解决了数据对不对，数据丢没丢的问题。但是数据的最终归属是为其他业务系统所用，业务系统拿到这些数据之后，通常还需要把数据和其对应的Schema匹配起来，才能进行业务逻辑处理。Schema 怎么办？MySQL数据库本身都不记录Schema变更的历史，而且MySQL的Schema定义是存储在MyISAM存储引擎中，是一个非事务引擎，变更处理过程中遇到宕机或者其他原因引起的Crash，可能就需要从备份重建实例了。而在实际的企业应用中，消费并对历史数据做计算是一种常态，所以要求我们也需要将Schema的历史版本记录下来，这样做到了数据和数据生成时Schema的准确一一对应。MySQL的Schema变更，是通过DDL语句来执行的，可以简单理解成SQL语句，比如：Create/Alter/Drop Table等。难道我们还要做SQL语法解析？显然这个代价太大，RDP中我们会将DDL回放到一个叫做Schema Store的MySQL实例中，这个数据库实例中只有Schema信息，没有业务数据。回放完成后，从Schema Store实例的information_schema库中查询出表的完整定义，然后以json格式记录到Schema Store实例的InnoDB表中，业务可以随时查询/简单解析处理。RDP中对Schema的变更历史处理，如下图所示：
上面提到了，由于MyISAM是非事务存储引擎，所以如果执行变更的过程中遭遇异常，通常都无法简单Rollback。RDP实现中采取了类似于两阶段提交的思想（不是两阶段提交的语义）来确保Schema的回放事务性，达到了Exactly Once语义，具体的方法在后面单独总结分享出来。
总结
总的来说，RDP：
通过给每一个事务分配单调递增的ID，达到了可以有效追踪一个事务在整个流转处理过程中状态的目的。
通过Binlog文件内部和跨文件的链式属性，形成了数据的上下文，达到了RDP系统自证和给其他系统提供他证能力的目的。
通过使用MySQL数据库本身来回放Schema变更的DDL，以及保证DDL执行的Exactly once，达到了历史数据和Schema完全一一对应的目的。
在较大规模的数据处理应用中，这些特性会有一定的积极作用。（避免数据的“死无对证”和团队间的“互撕”～～～）
