分片 (Sharding)

分片是分布式系统设计中的一个关键概念, 它与复制(Replication)一起, 构成了构建大规模系统的基础;

分片的本质与目标

分片的本质在于解决单一机器的限制和不可靠性问题,;

  • 复制(Replication): 通过复制数据来使机器可靠(reliable), 提高容错性;
  • 分片(Sharding): 通过分割数据来提高机器的容量限制 (raise the machine’s limits);
    • 分片适用于任何存储系统, 通常是通过将 键空间(key space) 分割到多个复制对(例如主/备份对)来实现;

Facebook 架构中的分片应用

  • 复制存储: 数据库在所有数据中心之间进行复制(replicated), 可以想象每个数据中心都包含一份完整的数据拷贝,;
  • 数据库分割: 数据库被分割成多个分片(shards), 每个数据元素都属于一个特定的分片,;
        - 领导者/跟随者: 每个分片都有一个数据中心作为其领导者(leader), 其余的中心是跟随者(followers);
        - 数量: 分片的数量通常远多于(many more)数据中心的数量;
        - 负载均衡: 分片被分配到不同的数据中心以实现    负载均衡(load balance)    ;
  • 写入和排序:
        - 对某一元素的更新操作总是发送给该元素所在分片的领导者(shard leader);
        - 分片领导者负责提供写入的排序(write ordering) 和 序列化(serialized);
        - sync: 写入 领导者的操作是同步(synchronous) 的, 当写入返回时, 对象就被持久存储了;
        - async: 领导者会将更新发送给跟随者, 但这是 异步(asynchronously) 进行的;

这种分片和异步复制的结合, 使得 Facebook 能够实现低延迟的写入, 但同时也意味着其他数据中心(跟随者)可能会看到旧值(old values), 这 不具备线性一致性;

Precedence Graph

类似于我们在数据库系统中看到的 precedence graph(优先图), Facebook 使用一种称为 Happens-Before Graph 的机制来跟踪操作的顺序;

  1. 创建顶点(Vertices): 为每一次操作(读或写)创建一个顶点;
  2. 添加边(Edges): 在操作之间添加边, 有两种情况:
    • 时序边: 如果操作 Op2 是在 Op1 的响应 被接收之后 才发出的, 则从 Op1 到 Op2 添加一条边;
    • 数据依赖边: 如果操作 Op2 读取了 Op1 写入的值, 则从 Op1 到 Op2 添加一条边;
  3. 简化图和检测违规: 在构建完包含所有时序和数据依赖关系的图之后, 需要进行简化和检测
    1. 合并 读取到写入: 将每一次读取操作合并到 产生该读取值的对应的写入 中;
    2. 检测循环: 如果在这个合并后的图结构中检测到 循环 (cycle), 则表示违反了线性一致性 (linearizability violation);

尽管 Facebook 的系统并未设计成线性一致性, 但通过这种检测发现, 只有极少数(0.0004%)的读取操作违反了线性一致性;然而, 这个事实本身意味着系统不提供线性一致性;

从车库的小引擎到世界第一互联网企业

名字起的大了点, 我们主要是想要讨论如何将小型的分布式系统逐步扩大到横跨全球的分布式集群:

要实现系统的真正扩展, 需要改变此前课程中的假设;

  • 此前的假设: 所有副本都拥有 完整 的状态(即每个副本都有每个键的值, 也就是 Replicated State);
  • 扩展需求: 为了扩展, 系统需要:
    1. 对状态进行分区 (Partition / split state);
    2. 将 partition 映射到 server 上;
    3. 需要有一种方法来 查找 哪个服务器拥有特定的分区;
      查找分区的方法可以通过数据本身描述, 通过增加复杂度的"定位服务", 或者简单地持续尝试直到找到为止;

那么如何分区存储呢?

  • 简单策略: 将学生的成绩单存储在 13 台服务器上, 根据姓氏首字母(如 A 或 B)进行分配;
  • 问题: 这种简单策略会导致负载在服务器之间出现倾斜(Skew in load), 可能造成高达 10 倍的差异 因为大多数数据都嵌入了倾斜或模式;

为了解决负载倾斜的问题, 分布式系统必须采用更复杂的 哈希 和分区策略(例如模哈希或一致性哈希, 这些内容将在后续介绍);

Distributed Hashing and Consistent Hashing

模哈希 (Modulo Hashing)

  1. 模哈希机制
    • 步骤: 对数据的键(key)应用加密哈希函数, 然后对服务器数量 N 取模 (modulo);
    • 存储位置: 将 (键, 值) 对存储在 hash(key) mod N 对应的服务器上;
    • 前提: 必须选择一个具有足够 熵(entropy) 的键;好的加密哈希函数会使得输出看起来是 随机的;
  2. 模哈希的缺陷 (N 变化的影响)
    • 模哈希的主要局限在于它 对系统成员变更的适应性极差:
    • N 变化: 当服务器数量 N 改变时(例如, 添加或移除一台机器), 几乎所有键的映射都会改变;
    • 大规模迁移: 这需要系统传输几乎所有(almost all)的值到新的服务器或新的位置, 导致数据迁移量巨大, 效率极低;

一致性哈希 (Consistent Hashing)

  1. 环形空间 与映射
    • 环形表示: 将哈希空间表示为一个圆环(circle),;
    • 服务器定位: 为每个服务器分配一个随机 ID(从键空间中抽取), 对该 ID 进行哈希, 并将其定位在环上;
    • 分片责任: 每个服务器负责其 predecessor 自身之间的键空间;
    • 键的映射: 要映射一个键到一个服务器, 首先对键进行哈希, 然后在环上执行读/写操作, 目标是其 successor;
  2. N 变更的 最小化迁移
    • 一致性哈希的关键优势在于, 当服务器数量发生变化时, 它能 最大限度地减少 状态迁移;
      • Join 服务器: 新的服务器只会从其后继那里 分割(splits) 一部分键空间;
      • Leave 服务器: 被移除服务器的分片将完全转移给其 successor 接管;
  3. 虚拟节点 (Virtual Nodes)
    • 为了更好地实现负载均衡和容错, 一致性哈希引入了虚拟节点(Virtual Nodes):
    • 机制: 每个服务器获取多个 (例如 v 个) 随机 ID, 每个 ID 对应一个虚拟节点;
    • 优势:
      • Load Balance: 更大的 v 值可以更好地均匀分布负载 (better load balancing), 使分片更可能均匀分布;
      • Adaptive to Different v: 可以根据服务器性能(异构性)调整 v 值(更强大的节点 v 值更高);
      • Fault Tolerance: 当一个服务器失败时, 其多个虚拟节点对应的分片会被不同的后继节点接管(v 个后继接管), 避免单一节点承受全部负载;

分布式哈希表 (Distributed Hash Tables, DHT)

一致性哈希定义了键到服务器的分配规则, 但要在大规模分布式系统中使用它, 客户端需要一种可扩展的方式来定位负责特定键的服务器;

Chord 协议 是 DHT 的一个典型示例, 它旨在实现对任何键的负责节点的可扩展查找, 目标是将查找时间从线性时间 O(N)O(N) 优化到对数时间 O(log⁡N)O(\log N);

查找面临的挑战(两个极端), 在设计查找机制时, 系统需要在两个极端之间进行权衡:
- 中心化 (Centralized) 方法: 某些机器知道系统中所有节点的位置 (O(N) 空间), 查找速度极快(O(1) 时间), 但维护集中状态困难;
- 这就是我们的 P4 shardmaster 的设计思路;
- 后继指针 (Successor Pointer) 方法: 每个节点只知道其后继节点 (O(1) 空间), 但查找可能需要遍历整个环, 最坏情况下时间复杂度为 O(N);

Chord 协议

Author: Ion Stoica (罗马尼亚超人), Robert Morris (388 蠕虫仙人), David Karger, M. Frans Kaashoek, and Hari Balakrishnan

指纹表 (Finger Tables)

Chord 协议通过引入 指纹表(Finger Tables) 来解决 O(N) 查找效率低的问题, 实现了类似二叉搜索的查找优化:

  • 基于排序: 键和节点在哈希环上都是有序的;
  • 指针数量: 每个节点维护 O(logN) 个指针指向其他节点;
  • 指纹表的构建: 每个节点的指纹表包含指向哈希环上特定距离的 successor 的 指针, 这些距离是 2 的幂次 (例如, 1/2, 1/4, 1/8…直到后继节点本身);
    • 节点 n 的第 i 个条目指向 hash(n)+2ihash(n)+2^i 的后继节点;

使用指纹表进行查找 (Lookup with Finger Tables)

查找过程通过递归地使用指纹表进行:

  1. 基本情况(Base Case): 如果目标键 k 属于当前节点的 successor 分片, 则返回该后继节点,;
  2. 归纳步骤(Inductive Step): 如果键 k 在其他范围内, 则在当前节点的指纹表中, 查找最远且不超过 k 的那个节点;然后将查找请求转发给该节点(即"在低端节点进行查找");
  3. 查找效率: 通过这种机制, 每一次查找都能消除哈希环上剩余范围的一半 (至少), 从而在最坏情况下将查找时间复杂度降低到 O(log⁡N)O(\log N);

然而, 即使是 O(log N) 的查找, 在节点数量巨大(如 100 万节点需要 20 跳)且节点跨地域部署时, 网络延迟的叠加仍然可能导致较长的实际查找时间 (例如 20 跳可能耗时 1 秒);

Example

  • 哈希空间位数 m = 8, 因此 ID 在环上属于区间 [0, 256)(首尾相接);
  • 当前存在的节点(按顺时针从小到大排序): {10, 42, 80, 120, 160, 200, 230}

定义: successor(x): 从 x 出发沿顺时针方向遇到的第一个节点;

Chord 在节点 n 的 finger table 定义:

  • finger[i] = successor(n + 2^i mod 2^m), 其中 i = 0..m-1;

计算 start_i = 80 + 2^i (mod 256), 再取 successor(start_i):

  • i = 0: start = 81 →\rightarrow successor(81) = 120
  • i = 1: start = 82 →\rightarrow successor(82) = 120
  • i = 2: start = 84 →\rightarrow successor(84) = 120
  • i = 3: start = 88 →\rightarrow successor(88) = 120
  • i = 4: start = 96 →\rightarrow successor(96) = 120
  • i = 5: start = 112 →\rightarrow successor(112) = 120
  • i = 6: start = 144 →\rightarrow successor(144) = 160
  • i = 7: start = 208 →\rightarrow successor(208) = 230

结论: 节点 80 的指表大多数项都指向 120, 较远的两项指向 160 和 230;

在当前节点 n:

  1. 令 s = successor(n);
  2. 如果 id $\in$ (n, s](顺时针区间), 则返回 s, 查找结束;
  3. 否则, 从 finger table 里选择 closest preceding finger(“最接近但不超过目标的指针”):
    • 选一个 finger 指向的节点 f, 满足 f $\in$ (n, id)(顺时针意义上在 n 和 id 之间), 并且 f 尽量靠近 id;
  4. 把请求转发到 f, 重复上述过程;

从节点 80 开始查找 id = 190

  • successor(80) = 120, 190 $\in$ (80, 120] 吗? False.
  • 选 closest preceding finger(在 (80,190) 里且最接近 190):
    • finger 候选: 120, 160 (230 超过 190, 不能选) -> 选择 160
  • 转发: 80 $\rightarrow$ 160, successor(160) = 200; 190 $\in$ (160, 200] 吗? True

因此答案: successor(190) = 200

路径: 80 $\rightarrow$ 160 $\rightarrow$ 200(命中)

备注: 和 “按 1/2, 1/4… 区间缩小” 的关系

  • finger table 的 2^i 让你拥有"指数级跨度"的跳点 (粗到细);
  • 但查找时不是显式找"落在哪个 1/2^k 区间", 而是选 closest preceding finger 来逼近目标;
  • 终止条件也不是"指针里直接命中目标", 而是当目标落入 (n, successor(n)] 这一段时直接返回 successor(n);

Dynamo 系统 (Amazon 的分布式键值存储))

  • Dynamo 核心原则:
    • 延迟: 关注 尾延迟(tail latency), 因为网页运行速度取决于最慢的响应, 中位数延迟不重要;
    • 无特殊节点: 系统中 没有 被指定为 “特殊” 的中心化节点;
    • 应用权衡: 赋予应用程序在一致性, 可用性和延迟 之间进行权衡的杠杆;
    • 因果追踪: 使用 causality 来追踪发散的更新版本;
    • 应用语义修复: 使用应用程序的语义知识来解决 conflict;

尾延迟 (First Contribution: Tail Latency)

Dynamo 系统的设计理念对分布式系统社区的一大贡献是强调了 尾延迟(Tail Latency) 的重要性, 而非传统的平均或中位数延迟;

  • 尾延迟的价值: 因此, Dynamo 强调应该关注尾延迟(tail latency), 例如延迟测量的 99.9% 分位数; 这对于提升用户体验更具实际意义;

Dynamo 中的一致性哈希和复制 (Consistent Hashing in Dynamo)

Dynamo 在一致性哈希(Consistent Hashing, CH) 的基础上添加了 replication 和 virtual 节点, 使其具备 容错性 和 耐久性;

  • 复制(Replication): 基础的分布式哈希表 (DHT) 通常假设单个服务器拥有键, 不具备容错性;Dynamo 通过为每个键配置 N 个后继节点作为副本 (N 是可配置参数) 来增加复制; 更大的 N 值可以提高数据的耐久性(durability)(即不丢失更新的概率);
  • 虚拟节点(Virtual Nodes): Dynamo 使用虚拟节点来解决负载均衡问题;它会跳过重复的服务器(即如果同一台服务器拥有多个虚拟节点), 以确保副本的独立性;

法定人数管理与权衡 (Quorum Management)

Dynamo 使用 quorum 来管理读写操作, 允许应用程序选择一致性模型:

  • Coordinator: 客户端将 Put 请求路由到 N 个后继节点中的一个, 该节点充当写入的协调器 (coordinator), 并将更新发送给其他 N−1 个副本;
  • 读写参数:
    • 写入操作等待 W 个响应才算成功(包括协调器自己的响应);
    • 读取操作等待 R 个响应才算成功(包括协调器自己的响应);
  • 权衡:
    • 最大化可用性/最小化延迟: 选择 较低 的 R 和 W 值;
    • 提供一致性: 如果设置 R+W>N, 则形成了 quorum (这里是鸽笼原理), 可以确保读取操作看到最近提交的写入
      • 不过这里由于 Dynamo 强调可用性, 支持并发写入, 所以可能会出现多个版本都是最新版的情况;

去中心化与 Gossip 协议

为了实现 “无特殊节点” 的原则, Dynamo 采用了 Gossip 协议来交换节点信息:

  • 机制: 服务器 定期随机选择一个 通信对象, 彼此交换服务器信息和虚拟节点 ID;
  • 优势: 该协议是完全去中心化的, 且随机选择 reaches saturation quickly;
  • 后果: 由于信息达到"完整"需要时间, N 个负责节点中的任何一个都可以作为协调器;这允许某些节点可能稍微过时, 并且由于没有单一节点来排序操作, 版本可能会发散(diverge);

为了解决这个版本发散的问题, Dynamo 使用了 Vector Clock 来追踪更新的因果关系, 也就是类似 Eventual Consistency 里的同步一致性机制;

向量时钟 (Vector Clocks in Dynamo)

由于版本可能发散(冲突), Dynamo 必须使用一种机制来追踪更新的因果关系, 以便解决冲突;它使用了向量时钟的一个变体;

  • 标准 VC 的问题: 标准的向量时钟的 vec 长度与系统中的服务器数量成正比, 在大规模系统中不具备可扩展性(not scalable);
  • Dynamo 的变体: Dynamo 使用了向量时钟的一个变体;它不存储系统中所有服务器的计数器, 而是存储一个 (协调器节点, 计数器) 对 的列表;
    • 内容: 该列表只追踪那些具有 非零值的节点;
    • 含义: 计数器记录了由该协调器节点管理的 更新数量;
    • 示例: 例如 [(A, 1), (B, 3), …], 其中 A 和 B 是更新的协调器节点;

通过上下文传输向量时钟 (Transmitting Vector Clocks via Context)

为了实现向量时钟的追踪, Dynamo 修改了传统的客户端接口, 要求客户端在读写操作中传递向量时钟信息:

  • Get 操作: Get(key) 操作返回 [value, vector clock], 即返回键的值和当前已知的向量时钟,;
  • Put 操作: 后续的 Put 操作需要将获取到的 vector clock 作为上下文发送回服务器: Put(key, value, vector clock);

向量时钟在副本中的用途主要有两个目的:
1. 解决冲突 (Resolving Conflicts): 向量时钟用于判断两个版本的关系:
- Causality: 如果向量时钟显示两个更新具有因果关系(即第二个更新是基于第一个更新的), 则第二个更新取代第一个更新, 第一个版本可以被丢弃(垃圾回收),;
- Concurrency Conflict: 如果向量时钟显示两个更新是并发的 (即它们之间没有因果关系), 则服务器必须保留所有冲突的值 (存储多个版本);
2. 垃圾回收 (Garbage Collecting Old Values): 通过确定一个版本被另一个版本取代后, 可以安全地将旧版本从存储中清除,;

冲突的解决方式 Resolve Conflicts

一旦识别出并发冲突, Dynamo 允许在不同的抽象层级进行解决:

  1. 系统层解决(System-Layer Resolution) 可以在系统层面上指定一个默认的解决策略, 例如**“最后写入者获胜”(last writer wins)**;

  2. 应用语义解决(Application Semantics) Dynamo 更鼓励基于应用程序语义的冲突解决, 借鉴了 Bayou 经验:

    • 客户端协调: 当多个版本存在并发冲突时, Dynamo 会将所有冲突的值返回给客户端;
    • 客户端决策: 客户端可以利用应用程序的语义知识来决定如何处理这些冲突, 例如对于购物车数据, 可以通过取商品数量的最大值(union) 来解决冲突;

  3. 服务器存储多个现实 (Multiple Realities)
    如果应用程序负责解决冲突, 那么服务器就必须集体存储所有冲突的值(即"多个现实");

  • 反熵机制的作用: 由于节点使用反熵机制学习到此前错过的操作, 如果这些错过的操作是并发发生的, 服务器必须保留所有这些版本, 直到客户端通过后续的 Put 操作解决了它们;

反熵 (Anti-Entropy)

节点的状态会随着时间而发散(diverge)或落后(out of date);反熵机制就是用来解决这种数据滞后问题的;

滞后的原因与传统系统的对比

  • out-of-date 滞后性: 在 Dynamo 中, 如果一个节点错过了一个更新, 并且此后没有针对该键的新值提交, 那么这个滞后的节点可能永远不会追赶上来 (catch up); Dynamo 并非为了最终更新而设计成最终一致性的;
  • Paxos 的解决方式: 相比之下, 像 Paxos 这样的系统要求集群就所有操作的顺序达成一致;落后的节点通常在处理它的下一个操作时就自动追赶上来了; Dynamo 没有等效的机制;
  • EC 解决方式: 通过定义一个 log 并且来交换 version vectors 来支持合并, 但是 Dynamo 没有 log

反熵机制的原理

为了确保所有节点最终收敛到一致的状态, Dynamo 节点会定期执行反熵操作:

  • 信息交换: 节点会定期交换关于它们共享分片的信息;
  • 原则上: 理想情况下, 它们会比较分片中所有的 (key, value, timestamp) 条目, 并复制任何一方有而另一方缺失的条目;

回忆一下我们在 p4 的实现过程中有一个自定义的 equals 函数, 也就是要求对所有的 slice 和 map 都要逐一对比每个元素是否相等, 那么这里就有问题了, 超大数据的情况下我们不应该完全逐个对比, 这会非常低效; 因为一组的不同 shard 之间的差异应该是非常 sparse 的, 逐个对比大部分时间是浪费的;

所以我们要借助 tree 结构来二分找差异性, 并且在每个子树中使用 hash 来快速对比;

Merkle Tree

  • 树按 key space 切分: 根覆盖整个空间(图里是 [0, 2^128) 这种大范围)
  • 左孩子覆盖前一半区间, 右孩子覆盖后一半区间
  • 继续递归切分, 直到叶子覆盖很小的区间(或覆盖一个 bucket)

每个树节点存一个 hash(摘要), 常见构造是:
- 叶子: 对该区间内的所有 (key,value)(按固定顺序)做一个摘要
- 内部节点: hash( left_hash || right_hash )

为了减少运算量, 一般从 root 开始比较, 相同就 prune 掉, 不同就继续往下比较

节点加入/离开可能非常昂贵 (Node Joins/Leaves Can Be Expensive)

  1. 昂贵的原因: 抽象耦合
    在现有的模型下, 如果要 join/leave, 可能包含的成本有: 读取所有的 kvStore 找到对应的 key 范围; 然后复制数据到新的结构并且发送给新的节点; 重新计算 Merkle Tree 等等; 而且注意 Dynamo 明确指出要重视 Tail Latency, 所以这样的行为不可接受;
    导致复杂性的根本原因是 Dynamo 使用单一 mechanism 来完成两个不同的 abstract:

    1. 键空间的 分区(Partitioning of the key space): 将数据状态分割成多个区块;
    2. 键到服务器的 放置(Placement of keys to servers): 将这些区块分配给特定的服务器;
    3. 由于这两个抽象是 耦合的, 因此每一次添加或移除节点, 都会同时修改分区和放置规则;

  2. 解决方案: 解耦分区和放置
    为了降低节点变更的成本和复杂性, 可以采取解耦分区和放置的策略:

  • 静态分区: 将键空间静态地划分为等大小的分片(shards); 类似于我们的 p4 实现的 16 个固定 shard;
  • 分片放置 (placement): 然后将这些分片 map 到 N 个虚拟节点上;
  • 分离存储和管理: 每个分片可以单独存储, 并维护自己的默克尔树 (shard level);
  • 迁移简化: 当节点加入或离开时, 系统只需要在节点之间交接分片/默克尔树, 而不是动态地重新计算哈希环上的分片范围;

Hinted Handoff

专门处理那些被写入到偏好列表 M 中, 但不属于该键的原始 N 个负责节点的更新; 这些更新被称为"远端更新"(far updates);

  1. 运作原理
    1. 临时存储: 当协调器发现一个属于 N 节点的更新无法写入到该节点时(例如节点暂时不可达), 它会将该更新转发给 M 列表中可用的, 但不属于 N 集合的某个节点 (例如节点 E);
    2. 标记原始目的地: 这个更新会在临时存储它的节点(节点 E)上被 标记(tagged) 上其"原始目的地"(original destination, 即 N 集合中不可达的那个节点 C);
    3. 转发数据: 当原始目的地节点(节点 C)再次可用时, 临时存储节点(节点 E)会将这个数据更新 转发(forwards) 回给节点 C;
    4. 遗忘: 一旦数据成功转发给原始目的地 C 并被确认接收, 临时存储节点 E 就可以 遗忘(forgotten) 这个数据了;
  2. 目标: 提示移交的目的是保证即使在节点瞬时故障期间, 写入操作也能完成(最大化可用性), 并且保证数据最终会被送到正确的, 负责持久存储它的 N 个节点之一(最大化耐久性);