Metastore

Megastore: Providing scalable, highly available storage for interactive services

在Bigtable上支持SQL，实现分布式数据库：

跨数据中心的多副本同步数据复制
支持为多数据表的字段建立Schema，且通过SQL接口访问
支持数据库的二级索引
支持数据库的事务

Megastore是Google迈向Spanner的一个里程碑，大部分特性基于实际场景做了取舍和妥协。更具实践意义

整体架构，如何从应用场景需求妥协而来
API设计、数据模型、事务和并发控制如何实现。如何利用Bigtable特性实现一个更高级数据库
数据复制机制，如何实现一个为跨数据中心的复制而优化的paxos算法

架构

互联网时代的数据库

早期的“可用性”往往局限于一个数据中心。GFS和Chubby中使用多个节点备份，但仍然在一个数据中心。但是，在可用性上一旦遇到水灾地震等自然灾害都无法保持可用；在性能上，跨地区访问可能造成很大的延迟。真正的高可用架构中，数据在多个数据中心同步，用户会就近访问最近的数据中心。一个数据中心出问题可访问其他的。这既保证了性能，有保证了可用性。

这样一个互联网层面数据库，用户量可能到达Facebook级别的几十亿。这个数据库要有非常强的水平拓展能力，能通过简单增加服务器就能服务更多用户。

Megastore的解决方案：

可用性：实现一个为远距离链接优化过的同步的、容错的日志复制器
伸缩性：通过把数据分区成大量的“小数据库”，每个都有独立进程进行同步复制的数据库日志，存放在每个副本的NoSQL数据存储中

复制、分区、数据本地化

Hive复用了MapReduce，Megastore复用了Bigtable。Megastore如何实现在多个数据中心的Bigtable间复制数据？

异步主从复制（Asynchronous Master/Slave）：两个问题：1.slave还没同步数据时master挂了，会丢失数据 2.在数据中心A写数据，在B读数据，那么刚刚写的数据会读不到，无法满足系统的“可线性化”
同步主从复制（Synchronous Master/Slave）：大部分系统解决方案
乐观复制（Optimistic Replication）：AP系统常用方案。数据可在任何副本写入，这个改动异步复制到其他副本。因为可就近写入所以性能好，但是因不知什么时候完成同步，所以只能是最终一致性。且无法实现事务，因两个并发写入的顺序无从判定，更谈不上隔离性和“可线性化”。

Megastore选择使用Paxos算法进行多数据中心的同步。不像之前的 Bigtable+Chubby，而是只采用 Paxos 确保只有一个 Master。Megastore 直接在多个数据中心里采用 Paxos 同步写入数据，是一个同步复制所有数据库日志但没有主从分区的系统。使用Paxos最大问题是性能：1.跨数据中心的网络时延 2.每次共识要过半节点确认

从业务需求到架构设计

在大型电商中，两个用户下单时间差距很小，那么订单顺序并不是那么重要。用户无法看见其他人订单，且业务上也无需分辨是谁先下的订单。可见业务上不一定需要全局“可线性化”，而只要一些业务上有关联的数据间能保障“可线性化”即可。不仅是可线性化，数据库的可串行化也是如此。例如一个单mysql的电商系统，两个用户同时下单。但是订单间不存在业务冲突，不会读写相同数据，此时可串行化的隔离式没有必要的。也就是可串行化要求不是全局的。

Megastore的提升性能的思路就在于无需全局的“可串行化”。引入了“实体组”（Entity Group）的概念，Megastore的数据分区也是按照实体组进行的。一个分区的实体组在多个数据中心间通过paxos进行数据同步读写。本质上，Megastore 将一个大数据表拆为多个独立的小数据表。每个小表在多个数据中心间通过Paxos进行同步复制，可在任何一个数据中心写入数据。但各个小表间没有“可线性化”和“可串行化”的保障，不保障先后顺序和隔离性。

图中行是数据分区，列是不同数据中心。

实体组是一个实体和这个实体下的一系列实体。例如电商中，每个用户是一个实体组，用户的所有订单、消息等都挂载在这个用户下。这些东西都打包在一起就是一个实体组，对这个实体组数据进行操作时可能产生冲突。Megastore 在每个实体组内支持一阶段的数据库事务，当存在跨实体组的操作时：1. 使用2PC事务，代价昂贵，是一个阻塞的、有单点的解决方案 2.抛弃事务，Megastore提供了异步消息机制。

异步消息机制如下：需要同时操作实体组A和B时，对一个实体组通过一阶段事务完成，然后通过Megastore提供的一个队列像实体组B发送一个消息。实体组B接收到消息后，可以原子地执行这个消息所做地改动。所以AB两边地改动都是事务性的，但两个操作没有共同组成同一个分布式事务。所以Megastore本质上没有实现数据库事务，而是实现了最终一致性。

以微信作为例子：

将每个微信号作为一个实体组
账号里收发地消息都挂载在这个实体组
发一条消息会影响两个实体组
但不需要保证发出消息和对方收到消息同时发生，所以可以使用Megastore的异步消息机制
先用一个一阶段事务，在发送者账号的实体组写入消息，再通过Megastore的异步消息机制向收件人的账号中发送一条写入消息请求。他的实体组收到异步消息后将消息事务性地写入自己的实体组

这个消息机制之所以可行，就是因为在实际应用层面，对于“可串行化”和“可线性化”的需求不是全局的，而是可分区的。只要保证发送者自己的聊天界面上消息是按顺序出现的，而不要求收件人和发件人间也是“可线性化”的。

小结

跨数据中心同步复制，让用户向最近的数据中心写入
支持为数据表建Schema、定义字段类型、支持SQL接口和二级索引、支持数据库事务
基于Bigtable和Paxos，实现了很多个分布式数据库的集合，而非一个“可线性化”的分布式数据库
支持一个实体组下的一阶段事务，括实体组的事务，要么使用代价高昂的二阶段提交，要么使用Megastore提供的异步消息机制
Megastore不是一个“透明”的分布式数据库，需要了解特性后对于自己的数据库表进行对应的适配性设计才能最大发挥功效

数据模型

Megastore虽然支持SQL形式接口，但应用中仍需自己对数据模型根据业务进行设计。它更像是对Bigtable上的应用层封装。它的数据模型是什么样？底层如何使用Bigtable存储？实体组层面事务是怎样？

实体组是什么

一系列会经常共同访问的数据，如用户和他的订单

-- 一个照片分享服务的 Schema
CREATE SCHEMA PhotoApp

CREATE TABLE User{
    required int64 user_id;
    required string name;
} PRIMARY KEY(user_id),ENTITY GROUP ROOT;

CREATE TABLE Photo {
    required int64 user_id;
    required int32 photo_id;
    required int64 time;
    required string full_url;
    optional string thumbnail_url;
    repeated string tag;
} PRIMARY KEY(user_id,photo_id),
    IN TABLE user,
    ENTITY GROUP KEY(user_id)REFERENCES User;

CREATE LOCAL INDEX PhotosByTime
    ON Photo(user_id,time);

CREATE GLOBAL INDEX PhotosByTag
    ON Photo(tag) STORING (thumbnail_url); -- 为全局索引定义了 STORING 字段，并指向Photo中的 thumbnail_url 字段

定义了一个PhotoApp的Schema，类似数据库的库。定义了一张叫User的表，定义 user_id 是主键，并定义了这个表是实体组（Entity Group）的一个根（Root），一条User表的记录代表一个用户。定义了Photo表，其中主键是两个字段的组合。Photo表通过 user_id 字段挂载到User表上。这里类似数据库的外键，可以层层挂载。之后还建了两个索引：本地索引、全局索引。

这个Schema类似Thrift和Protobuf的定义文件，有类型、required、optional、repeated字段。大数据系统常见方案：为了减少要跨越特定服务器进行Join，直接支持嵌套的List类型字段。

实体组的数据布局

示例数据在Bigtable中存储的数据布局如图，PhotoApp中User和Photo都存在同一个Bigtable中，绿色部分是User表，蓝色是Photo表。使用User表的user_id作为行键，对于Photo表使用 user_id 和 photo_id 组合作为行键。因为Bigtable中数据按行键连续排列，所以同一个User下的Photo数据会连续存储在对应User后。Bigtable中数据按照行键分区和连续存储，通过行键读取数据时有Block Cache，也就是读取整个SSTable的block，这里就包含了当前行键的前后连续数据。所以读取一个User记录时有很高概率读取相关的Photo记录，此时无需再访问对应硬盘。这种数据布局被称为对Key进行预Join（Pre-Joining with Keys）。

为避免热点问题，支持对数据表添加一个SCATTER参数，这使所有行键带上Root实体记录的Key的哈希值。这样，虽然同一实体组的数据还是连续排列，但同一张表的两个连续实体组的Root记录的Key不一定存在一个服务器。

数据库的每个列直接使用Bigtable的Column，且Megastore这样混合一个实体组里的多个表的结构，是非常适合Bigtable的。因为Bigtable是稀疏的，对不存在的列无需存储，也不会占用空间。

Megastore 的索引

索引分为两类：

本地索引：直接存储在实体组内部，在确定哪一个实体组是寻找具体记录位置
例如例子数据中PhotosByTime索引，需要通过user_id和time两个字段才能访问。就是先通过user_id确定实体组再查询数据
全局索引：无需确定实体组，但是更新不是实时的。异步更新导致弱一致性。

索引优化

Megastore支持在索引中存储数据
传统数据库索引中只能获得记录的主键，需要查询数据要拿着主键再查询一次。Megatore中可通过STORING语句指定索引中存储对应数据的某个字段的值，这样检索索引即可拿到值。这在分布式数据库中很重要，因为不同于单机数据库，索引和数据不在一个节点，多一次网络往返会有很大延迟
支持为 repeated 字段建立索引（Repeated Indexes）
例如数据中PhotoByTag字段，对应的索引字段是Photo实体里被repeated修饰的tag字段。Megastore为每个tag记录一条索引，这样通过索引反向查询到某个tag关联的所有Photo记录。如果为每个repeated字段单独建一张表，无论是独立的表还是像实体组一样挂载到Root表上，都很浪费空间且将结构复杂化。
提供了对内联索引（Inline Indexes）的支持
帮助父实体快速访问子实体的某些字段。如PhotoByTime原本是Photo实体的索引，变成User实体的内联索引。这样User表相当于多了一个repeated的虚拟字段，这个List里都存放两个信息：PhotoByTime里的time信息，这个time对应的是Photos中的哪一条记录。
这样父实体添加一个虚拟字段后，查询子实体时直接在父实体中即可完成，而无需查询具体的索引数据。

索引实现

每条索引都是一行数据存在Bigtable中。行键就是建立索引的字段和索引到数据的主键的集合。

CREATE LOCAL INDEX PhotoByTime
    ON Photo(user_id, time);

此索引由两个字段构成
索引的是Photo表，所以Photo表的主键就是 user_id 和 photo_id
索引这一行的行键就是 ((user_id, time)), (user_id, photo_id)) 的组合
如果索引指向一个repeated字段，如 tags。每个tag都有一行数据，如三个tag，分别是 [tag1, tag2, tag3]，索引也会有三条记录 (tag1, (user_id, time)), (tag2, (user_id, time)), (tag3, (user_id, time))

Megastore 的事务和隔离性

同一实体组下的所有数据行可看成一个抽象的小数据库，在之上Megastore也支持“可串行化”的ACID语义。采用MVCC（Multiversion Concurrency Control 多版本并发控制）实现隔离性。数据库有多个历史版本，每次请求那最新的快照，整个事务提交时会检查当前数据库里数据的最新版本，是否和拿到的快照版本一致。如果一致则提交成功并更新版本。当有两个并发的数据库事务读写同一份数据时，先提交的事务会成功，后提交的会因为最新版本数据变了而失败。当有一个事务正在提交或写入到一半，另一个读取事务的请求不会读到已经写的一般数据，而是读到上一个最新版本的快照。

单条记录有多个版本，可线性化要求提交时检查是否有其他事务已经更新过当前事务读写的数据

Bigtable天然会存储多个版本的数据，每次写入都是添加新版本，这和MVCC很匹配。Megastore的一个实体组可能包含多行数据，使用时间戳实现基于MVCC的事务和隔离。提交事务时只要指定时间戳，读取时直接读最新的。如果事务执行一般，一个实体组中部分数据更新，此时不会读到更新一半的数据。根据时间戳，提供了 current、snapshot、inconsistent 三种模式：

current：读取最新版本数据。确认已提交的事务全部成功完成，然后读取最新事务对应的时间戳的数据版本
snapshot：不会等待当前是否有已完成提交的事务应用，而是直接返回上一个完全应用的事务对应的数据版本
inconsistent：完全忽视事务系统的日志信息，直接读取最新数据，这时会读取“不一致”的数据，因为可能读取到提交了一半的事务，可能不同行不同版本的数据

Metastore是在Bigtable之外添加了一个独立的事务系统。就是一个复制日志的状态机。事务提交分下面5个步骤：

读（Read）：先获取时间戳，最后一次提交的事务的日志位置
应用层逻辑（Application Logic）：从Bigtable读数据，并把所有需要的写操作收集到一条日志记录（log entry）中
提交事务（Commit）：通过Paxos算法在数据中心副本间达成一致，把这个日志追加到最后。类似预写日志（WAL）
应用事务（Apply）：实际对实体和索引做修改，写入Bigtable。类似MemTable+SSTable
清理工作（Clean UP）：删除不需要数据

第3、4步间存在时间差，导致区分不同模式，预写日志已完成，但数据没更新到Bigtable：

current：等数据更新完成，再获取这个最新数据
snapshot：不等待完成，直接读上一个版本数据

所有的读都使用current读，可保障“可线性化”，但时延和性能很糟糕。

并发写：第一步获取最新日志位置，在第3步竞争写入同一个日志位置，但只有一个成功，失败的那个从头再来。

数据复制机制

解决跨数据中心的延时问题，利用了单master性能优点，又不用承担master可用性缺点。

Chubby 是粗粒度锁

Chubby的整个Paxos集群要选出一个master，所有数据读写通过master进行，其他节点只是容错。造成性能瓶颈
master本质是单点，出错时需要超时检测才能等新的master选出，然后才能提供服务。会有一段不可用时间

作为粗粒度锁，Chubby只要管理极少变化的元数据，不需要管理高并发的数据库事务请求。这种单master实现的Paxos算法不适合Megastore，而且还有“网络时延”。时延在跨数据中心复制中非常重要。

Megastore应该让提案可从任何节点发起，即可发到离他最近的数据中心。

单master的paxos算法也有好处：两阶段的 Prepare-Accept 请求，可在 Accept 请求时加上下一次的 Prepare 请求。因为所有请求都先到Master再发起提案，所以达成共识过程中很少有冲突。单master可减少Paxos算法网络开销。

单master的优点：合并请求减少网络开销；缺点：单节点写性能瓶颈。处理方法：基于Leader的Paxos算法。

两个保障

Megastore 支持三种读方式：current、snapshot、inconsistent。为了可线性化要实现current，要做出两个保障：

每次读都要能观察到最后一次确认被写入
一旦一个写入被观察，所有未来的读都能观察到这个写入
在写入提交前后续数据可取可以读到这个未提交写入，这样才能保障数据的“可线性化”

数据的快速度

为了快速读，要从本地副本读取数据。但paxos不是2PC复制算法，而是“共识”算法。只要超过一半的节点投票就可达成共识，某个节点可能没参与共识，即使达成共识将日志写入本地也要一段时间。这意味某个节点不一定能获得最新数据。所以Megastore在每个数据中心都引入一个协同服务器（Coordinator Server），用于追踪当前数据中心副本中最新实体组成的集合。如果实体组在这个集合只要从本地读取即可，否则要“追赶共识”（catch up）

实际上Megastore读过程比“从本地副本读”要复杂：

1. 查询本地协同服务器，查看实体组是否最新

2. 根据查询结果，判定从本地副本还是其他数据中心的副本找到最新的事务日志位置

日志位置类似自增的事务日志ID。因为事务日志写到Bigtable的一张表里，因为Bigtable支持单行事务，所以事务日志作为一行数据写道Bigtable也是一个原子提交。Bigtable的数据按照行键连续存储，也很适合事务日志这种追加写的特性：

如果协同服务器告诉我们，本地的实体组就是最新的那就从本地副本，拿到最新的日志位置以及时间戳。在实践当中，Megastore不会等待查询到本地是不是最新版本再来启动这个查询。而是通过并行查询的方式来缩短网络延时，即使本地不是最新版本，也无非浪费一次Bigtable的数据读取而已。
如果本地副本不是最新的，那么会向Paxos其他的节点发起请求得到最新日志位置。根据多数意见知道此时最新的事务日志的位置。然后，挑一个响应最及时，或者拥有最新更新的副本，从它那里来开始“追赶共识”。因为本地节点往往是响应最快的，所以从本地副本去“追赶共识”，往往也会是一种常用策略，但这不是必然选择。

3. “追赶共识”的过程，确定同步副本的目标后：

使用paxos算法的noop操作确定从哪个日志位置开始同步
在这个副本的Bigtable里，顺序应用这些事务日志，让这个副本的Bigtable追赶上整个分布式系统的“共识”
在追赶共识过程中失败就换个节点继续

4. 如果是向本地副本发起追赶，那么追赶完成意味着达成最新状态。他向本地协同服务器发起一个Validate消息，使其知道这个实体组数据是最新，本地副本无需等待回复，失败也只是下次再同步一下。

5. 根据用了哪个“副本”来追赶共识，通过拿到的日志位置和时间戳，向它的Bigtable要数据。此时副本不可用，要再找一个副本，从第三步的“追赶共识”开始重复一遍。

在多个数据中心的Paxos集群大概率会正常参与整个共识过程，故障和执行上述复杂过程是低概率。步骤 1 2.1 5 是用于获取数据，步骤 2.2 3.4 都是为了容错的“可线性化”的表现。

数据的快速写

读时尽可能从本地读，写时使用基于Leader的实现方式。将Accept和后一次的Prepare请求合并，当Leader故障时退回到原始Paxos算法即可。

为了确保“可线性化”，在写之前都会先“读”一次数据，确保下一次事务日志位置、最后一次写入数据的时间戳，以及哪个副本在上次Paxos算法被确定为Leader。通过每次提交事务时的"写"确定Leader，每次事务写入的处理事务内容还有对Leader节点的“提名”，事务写入成功后提名就通过，下次Paxos算法的Leader就确定。

确认下一个事务日志位置，Leader是哪个节点后，可以写入数据：

Accept Leader 阶段：客户端向Leader发起Accept请求，且设定为第0轮Accept请求，若被Leader接受，会跳到第3步，向所有副本发起Accept请求达成共识
第1步失败就执行普通的Paxos算法流程，向所有副本发起一个Prepare请求。附带的提案号就是正常的Paxos算法提案编号，从1开始，用当前客户端见过的最大编号+1
Accept 阶段：所有副本接受客户端的提案，没有多数通过则进入第2步，重新走Prepare-Accept 过程
Accept成功后，向所有没Accept最新值的副本发起一个Invalidate请求，确保协同者服务器把对应实体组从Validate的集合中去除掉
Apply 阶段：客户端让尽可能多的副本将实际修改应用到数据库。如果发现要应用的数据和实际提案的数据不同，返回一个冲突的报错（conflict error）

Megastore的Paxos实现事务和之前的Paxos算法和数据库事务有点不同：

Paxos 算法层面：无论哪个客户端的数据中心发起事务都会先向Leader提交，和单master类似方法解决并发冲突，无需多轮Paxos协商。Leader挂了直接退回到多轮协商Paxos算法，避免了单master的可用性问题
Megastore 数据模型：每个实体组都是“数据库”，针对同一个实体组Accept请求指定下一次的Leader。同一个实体组往往是同一段时间、同一地域内多次写入。每次提名下一个Leader时，只要提名此时相应最快的最近副本即可，这样大概率下次数据写入也是“就近”的。
例如User实体组有多个Photo子实体，因为一个用户是一个实体组，一定时间内图片上传都是在相近区域，自然Megastore可根据前一次写入时离哪个数据中心近，就提名哪个数据中心为Leader。
这个策略是和Megastore的数据模型、开发者设计的数据表结构高度相关的。如一个订单实体，Order作为订单组，订单可能来自全国各地，大量跨地域事务提交非常影响系统性能。
数据库事务层面：传统数据库中提交点（commit point）和可见点（visibility point）相同；但Megastore中在事务提交后不是所有副本都接受了这个最新事务的“值”，也不是所有协同服务器都将对应实体组标记为“失效”。当这两者完成后才能确保下次读取副本是一定能读到实体组最新版数据

“写入可能在被确认之前就被观察到”：在写入的第三步，部分节点写入数据，部分节点没有写入。此时读取不同节点可能读取到或读取不到最新数据，即事务确认（acknowledge）的过程没有完成。

系统整体架构

事务提交成功和实际存放的Bigtable里可见是两个步骤，这时就需要协同服务器。Megastore的事务日志和数据库数据都是存储在数据中心的Bigtable，但不知道本地的事务日志和数据是不是最新版本。这要通过协同服务器维护。好在协同服务器和数据存在内存，且只要维护本地数据中心的实体组是否是最新这一个状态。即使节点故障，也只要重启一个新节点，数据可通过Paxos投票选出进行恢复。因为简单，所以协同服务器比Bigtable更稳定。

Megastore 中协同服务器的容错如此：

服务启动时取Chubby中获得特定锁
要处理请求，协同服务器要持有过半的锁
一旦因网络故障导致它不再持有过半的锁，那么退回到保守的默认状态，也就是认为所有的本地实体组都是过时的
后面的数据都要通过Paxos投票方式请求远程副本

也就是协同服务器决定网络分区了，那么就放弃自己维护的本地状态，退回到原始的Paxos多数投票策略。

横向看：

每个数据中心都有一个应用服务器，所有外部请求都先到应用服务器在有它代理进行数据库操作
事务日志和数据等都是存储在底层的Bigtable中
在中间层，Megastore有两类服务器：协同服务器，维护本地数据是否最新版本；复制服务器（Replication Server）。当有数据写入请求时，写到本地数据中心的直接写Bigtable，如果是写远端其他数据中心，则是发送给那个数据中心的复制服务器。其本质上是一个代理。此外复制服务器还定期扫描本地副本，将因网络或硬件故障而没完整写入的数据库事务同步到最新

为了提升效率，每个数据中心的副本还分为三类：

完全副本（Full Replica）拥有所有服务
见证者副本（Witness Replica）只参与投票和记录日志，不保留实际数据，也无法用于查询。如果完整副本较少，如只有北京上海两个数据中心，这无法完成Paxos需要的最少三个节点的投票机制。这时可随便选一个见证者副本确保可完成“多数通过”投票机制
只读副本（Read-Only Replica）不参与投票，只包含完整数据库数据，是一个异步的数据备份。在无需保障“可线性化”时，直接读取这个数据，在读多写少时很有用