Google的三架马车: GFS, MapReduce, BigTable

一、GFS(Google File System)

1.1 GFS简述

GFS是Google在2003年前后创建的可扩展分布式文件系统，用来满足Google不断扩展的数据处理需求。为大型网络和连接的节点提供容错、可靠性、可扩展性、可用性和性能。GFS由多个由低成本商品硬件组件构建的存储系统组成(使用了集群的概念，联合了多个硬件)，它经过优化以适应Google的不同数据使用和存储需求，例如其搜索引擎会生成大量必须存储的数据。

1.2 GFS的特点

• 由许多经常出现问题的廉价硬件构建而成。但是能不断地监控自己，并在常规的基础上及时发现、容忍和恢复组件故障。
• 存储着非常多的大文件，对大文件进行了管理优化。
• 对大文件顺序流式读写操作进行了优化以及对小文件随机读取进行了批处理和排序等优化。
• 任何一个客户端都可以访问到存储在不同硬件上的某个文件，支持对文件的并行处理(同时读，同时写，边读边写)，因此原子性操作以及允许把一个大文件切片存在不同的硬件上也是必不可少的。
• 高持续带宽比低延迟更重要。因为目标应用大多偏重于处理大批量、高速率的数据，而很少有针对单个读或写的严格响应时间要求。

1.3 GFS的架构

主要是以下几个特点：

• 由单个主服务器和多个块服务器组成，由多个客户端访问。
• 文件被划分为固定大小的块，一般为64MB。
• 为了保证可靠性，每个组块在多个组块服务器上进行复制，默认情况下会复制三份。
• 主服务器维护所有文件系统元数据。这包括命名空间、访问控制信息、文件到组块的映射以及组块的当前位置。
• 客户端与主服务器交互进行元数据操作，但所有的数据承载通信都直接交给组块服务器。

可以看到客户端向主服务器发送文件访问请求，而主服务器只返回了文件句柄和所在的服务器位置，然后客户端拿着这个数据去向返回过来的数据块服务器地址要数据。

1.3.1 单主模式

• 优点：极大地简化了GFS设计，因为单一主机可以使用全局知识做出复杂的块放置和复制决策。

• 缺点：

  • 单点故障。需要定期的将关键元数据检查点放到非易失性存储中。
  • 可能是系统的性能瓶颈点。必须尽量减少主服务器参与读写的操作。

  对于第二点，在架构图中我们可以发现，客户端从来不通过主服务器进行读取和写入文件数据。客户端会先询问主服务器他应该联系哪个块服务器，在限定的时间内缓存此信息(指应该联系的块服务器的信息)，并直接与块服务器交互以进行读写操作。

1.3.2 块大小

块大小选择了64MB，它比典型的文件系统块大得多。每个数据块副本以普通Linux文件的形式存储在一个数据块服务器上，并根据需要进行扩展。

• 优点：

  • 减少了客户端与主端交互的需求，因为对同一块进行读写操作只需要向主端发出一个初始请求即可获得块位置信息。
  • 在一个大的数据块上，客户端更有可能在给定的数据块上执行许多操作，因此可以通过在较长时间内保持与数据块服务器的持久TCP连接来减少网络开销。
  • 减少了存储在主节点上的元数据的大小。这使得我们可以将元数据保存在内存中。

• 缺点：一个大块量度，即使有懒惰的空间分配，也有它的缺点。

  • 小文件的碎片化存储浪费空间。
  • 如果有多个客户端访问同一个文件，那么存储这些数据块的数据块服务器可能会有多个并发请求导致请求过载。

  对于第二点，可以通过将这些可执行文件以更多的复制因子存储，并使批处理系统错开应用程序的启动时间来解决这个问题。还有一个潜在的长期解决方案是允许客户在这种情况下从其他客户那里读取数据。

1.3.3 元数据

元数据存储在主内存中(块大小的优点提到了)，而数据存储在块服务器中。这使得主服务器操作非常快，并且还允许主服务器在后台通过其整个状态有效地执行定期扫描。周期性扫描用于实现块垃圾收集、块迁移等。

主服务器存储以下三种元数据：

• 文件和Chunk的命名空间。
• 文件和Chunk的对应关系。
• 每个Chunk副本的存放地点。

请注意，前两种类型的元数据通过将更改记录存储到主服务器本地磁盘上的操作日志中来保持持久性，并定期复制到远程机器。但是，主服务器不会持久存储块位置信息。相反，它会在主服务器启动时以及当有块服务器加入集群时询问每个块服务器关于它的块数据。这是因为块服务器是块位置和主要状态信息的权威数据源。

1.3.4 一致性模型

GFS有一个宽松的一致性模型，也可以称之为弱一致性模型，它很好地支持高度分布式应用，但仍然保持相对简单和高效的实现。

论文中作者描述了文件区域的两种状态：

• 如果所有客户端，无论从哪个副本读取，读到的数据都一样，那么我们认为文件region是“一致的”；
• 如果对文件的数据修改之后，region是一致的，并且客户端能够看到写入操作全部的内容，那么这个region是“已定义的”。

其他操作：

• 原子操作：
  GFS提供了一些原子操作。文件命名，空间修改例如文件创建都是原子性的，并且由主服务器专门处理。命名空间锁(Namespace locking)保证了原子性，主服务器的操作日志定义了这些操作的全局顺序。
• 记录追加：
  与常规的写入操作或者最佳操作相比，GFS提供了记录追加操作，即使存在并发冲突的情况下，也能够保证要追加的数据至少以原子操作的方式执行一次，但以GFS选择的偏移量(offset)。GFS将会返回这个追加数据的偏移量给客户端(并不能避免冲突，但是让冲突变得可容忍)。

1.4 总结

上面的内容就是GFS的核心内容了，但是并没有很深入的探讨同时也还有很多内容没有讲到，比如主服务器的垃圾回收机制等等，但是理解其核心思想就已经可以了。

二、MapReduce

2.1 MapReduce简述

MapReduce是一个分布式、并行处理的计算框架。
MapReduce把任务分为Map阶段和Reduce阶段。开发人员使用存储在DFS中数据(可实现快速存储)，编写分布式计算平台的MapReduce任务。由于MapReduce工作原理以及DFS的特性，平台能以并行的方式访问数据，从而实现快速访问数据。

2.2 MapReduce编程模型

• Map函数由用户编写，取一个输入对，并产生一组中间键/值对。MapReduce库将所有与相同中间键key_i相关联的中间值组合在一起，并将其传递给Reduce函数。
• Reduce函数也由用户编写，接受一个中间键key_i和该键的一组值。它将这些值合并在一起，形成一个可能更小的值集合。通常每次Reduce调用只产生0或1个输出值。中间值通过迭代器提供给用户的reduce函数。这使得我们可以处理太大而无法在内存中拟合的值列表。

这里给出基于java的实现单词计数的写法示例：

public static class TokenizerMapper
        extends Mapper<LongWritable, Text, Text, IntWritable> {

    private final static IntWritable one = new IntWritable(1);
    private Text word = new Text();

    public void map(LongWritable key, Text value, Context context) 
            throws IOException, InterruptedException {
        // 将输入行转换为小写，并移除标点符号，只保留字母和空格
        String line = value.toString().toLowerCase().replaceAll("[^a-zA-Z\\s]", "");
        // 使用 StringTokenizer 按空格分割单词
        StringTokenizer tokenizer = new StringTokenizer(line);
        while (tokenizer.hasMoreTokens()) {
            String token = tokenizer.nextToken().trim();
            // 只处理非空单词
            if (!token.isEmpty()) {
                word.set(token);
                context.write(word, one);
            }
        }
    }
}

public static class IntSumReducer
        extends Reducer<Text, IntWritable, Text, IntWritable> {
        
    private IntWritable result = new IntWritable();

    public void reduce(Text key, Iterable<IntWritable> values, Context context) 
            throws IOException, InterruptedException {
        int sum = 0; 
        // 累加同一个单词的所有计数
        for (IntWritable value : values) {
            sum += value.get();
        }
        result.set(sum);
        context.write(key, result);
    }
}

Map和Reduce函数对应的输入和输出如下(list表示的是返回零到多条数据)：

函数	输入	输出
map	(k1, v1)	list(k2, v2)
reduce	(k2, list(v2))	list(v2)

2.3 执行过程

MapReduce应用程序执行过程：

分为以下七步：

• 用户程序中z的MapReduce库首先将输入文件分割成M块，通常为16兆字节到64兆字节每块(由用户通过可选参数进行控制)。然后，它在集群机器上启动许多程序的副本。
• 程序的副本之一是特殊的–主程序。其余为由主程序分配工作的工人程序。有M个map任务和R个reduce任务需要分配。主程序挑选闲置工人程序，并为每个人分配一个map任务或reduce任务。
• 被分配map任务的工人程序读取相应输入分片的内容。它从输入数据中解析出键/值对，并将每个键/值对传递给用户自定义的Map函数。Map函数产生的中间键/值对在内存中缓冲。
• 当主站通知reduce工人程序位置时，它使用远程调用从map工人程序的本地磁盘读取缓冲数据。当reduce工人程序读取所有中间数据时，它根据中间键对其进行排序，从而将相同键的所有出现归为一组。排序是需要的，因为通常许多不同的键映射到相同的reduce任务。如果中间数据量太大，不适合在内存中使用，则使用外部排序。这一段也是shuffle过程。
• reduce工人程序对排序后的中间数据进行迭代，对于遇到的每个唯一中间键，将该键和对应的中间值集合传递给用户的Reduce函数。Reduce函数的输出被添加到这个reduce分区的最终输出文件中。
• 当所有map任务和reduce任务都已完成时，主程序唤醒用户程序。此时，用户程序中的MapReduce调用返回给用户代码。

成功完成后，MapReduce的输出执行可用在R输出文件(每个reduce任务一个,文件名由用户指定)中。通常，用户不需要将这些R输出文件组合成一个文件–他们经常将这些文件作为输入传递给另一个MapReduce调用，或者使用它们从另一个分布式应用中处理被分割成多个文件的输入。

2.4 总结

同样只是简单地讲了核心思想以及流程，很多东西也并没有涉及到，包括shuffle的优化，数据传输以及任务粒度等。想尝试一下MapReduce的话，可以使用Hadoop平台完成简单的WordCount任务，Map和Reduce函数也已经给出。

三、BigTable

3.1 BigTable简述

BigTable是一个用于管理结构化数据的分布式存储系统，其设计目标是扩展到非常大的规模：跨越数千个商品服务器的数百亿字节的数据。它实现了几个目标：广泛适用性、可扩展性、高性能和高可用性。

设计动机：

• 需要存储的数据种类繁多：包括URL、网页内容、用户的个性化设置在内的数据都是Google需要经常处理的。
• 需要存储的数据种类繁多海量的服务请求：Google运行着目前世界上最繁忙的系统，它每时每刻处理的客户服务请求数量是普通的系统根本无法承受的。
• 商用数据库无法满足需求：一方面现有商用数据库的设计着眼点在于其通用性，另一方面对于底层系统的完全掌控会给后期的系统维护、升级带来极大的便利。

3.2 数据模型

BigTable是一个稀疏的、分布式的、持久的多维排序映射。map由row_key、column_key和timestamp索引；映射中的每个值都是一个字节的无解释的数组(只负责存储，不负责读懂)。

(row:string, column:string, time:int64) → string

下图是一个存储网页的Webtable：

• row是url的倒转，比如www.google.com会存为com.google.www，这样的目的是前面的www大家都一样，而且子域名就会主域名靠一起，方便聚簇。
• 有多列，其中contents:列存储网页html内容。anchor:列存储指向这个页面的anchor文字，比如cnnsi.com和my.look.ca有指向www.cnn.com的anchor，就如下图所示存储。
• 每一个anchor cell只有一个版本，但是content cell有三个版本为t3，t5，t6。

3.2.1 Row

• 行key是表的主键，可以是任意字符串，最大为64kb，在单行的读写都是原子的，使客户端在同一行存在并发更新的情况下更容易推断系统的行为。由于读写总是通过行键，这样的数据库也叫做KV数据库。
• BigTable按行key对数据进行排序，行范围动态分区，每个行的范围被称为tablet，是分布式和负载均衡的单位。因此，短范围的行读取是高效率的，通常只需要与少量的机器进行通信。客户可以通过选择他们的行键来利用这一特性，从而使他们的数据访问获得良好的局部性。

3.2.2 Column Families

• 列族是访问控制的基本单位，同一列族内的所有数据通常类型相同(我们会把同一列族的数据一起压缩)。必须先创建列族，然后才能在该族下使用任何列键；列族一旦创建，其内部任何列键都可随时使用，也就是说一行的某个列族下可能会出现空值，这也就是说为什么BigTable是稀疏的。设计时要求一个表中的列族数量很少(最多几百个)，且运行期间极少变更。相比之下，表的列数可以无限。
• 列键的命名语法为：family:qualifier。列族名称必须是可打印字符，限定符则可以是任意字符串。以Webtable为例，可以用language当做family，另一种是可以用anchor来当做family(如上图所示)，每个列key是一个anchor，qualifier是指向该url的网址，内容是链接文本。
• 访问控制以及磁盘、内存的计量都在列族级别进行。在Webtable中，这些控制让我们能够管理多种应用：一些负责添加新基础数据，一些读取基础数据并生成派生列族，还有一些只能查看已有数据(甚至可能因隐私原因无法查看所有列族)。比如BigTable的开源实现HBase，每一个列族的数据存在同一个HFile文件下。

3.2.4 Timestamp

• BigTable中的每个单元格可以包含同一数据的多个版本；这些版本都是以时间戳为索引的，时间戳为64位整数。不同版本以递减的形式存储，以便可以首先读取最新版本。
• 为了减少手动管理多版本数据的麻烦，BigTable给每个列族提供了两种自动垃圾回收(GC)策略，让系统自己帮用户清理旧版本：
  • 只保留最近n个版本(老版本一旦超过n个就自动删除)。
  • 只保留“足够新”的版本(n天前写入的任何版本都会被自动清理掉)。

3.3 API

BigTable API提供了创建和删除表和列族的功能。它还提供了更改簇、表和列族元数据的功能，如访问控制权限。

• 客户端应用程序可以在大表中写入或删除值，从单个行中查找值，或者在表中的数据子集上进行迭代。

  // Open the table
  Table *T = OpenOrDie("/bigtable/web/webtable");
  
  // Write a new anchor and delete an old anchor
  RowMutation r1(T, "com.cnn.www");
  r1.Set("anchor:www.c-span.org", "CNN");
  r1.Delete("anchor:www.abc.com");
  Operation op;
  Apply(&op, &r1);

• 客户端可以在多个列族上进行迭代，并且有几种机制来限制扫描产生的行、列和时间戳。

  Scanner scanner(T);
  ScanStream *stream; stream = scanner.FetchColumnFamily("anchor");
  stream->SetReturnAllVersions(); scanner.Lookup("com.cnn.www");
  for (; !stream->Done(); stream->Next()) {
    printf("%s %s %lld %s\n",
    scanner.RowName(),
    stream->ColumnName(),
    stream->MicroTimestamp(),
    stream->Value());
  }

3.4 构建块

• BigTable使用分布式的GFS来存储日志和数据文件，依赖于集群管理系统，用于调度作业、管理共享机器上的资源、处理机器故障和监控机器状态。

• BigTable数据格式为SSTable，SSTable提供了一个不可变的、有序的从键到值的不可变映射。每个SSTable包含一个块(通常每个块的大小为64KB)，块索引(存储在SSTable的末尾)用于定位块。SSTable打开时索引被加载到内存中，查找时先内存二分查找索引找到块，再从磁盘一次读取指定的块(也可以把整个SSTable加载到内存，这样就不涉及到磁盘了)。

• BigTable依赖于一种称为Chubby的高可用性和持久性的分布式锁服务。Chubby包含了5个副本，其中一个被选为master并提供request服务，采用5节点部署，多数派存活即可服务；用Paxos选主并同步小文件内容与锁状态。Bigtable主要通过Chubby完成以下5个任务：

  • Master选举：同一时刻最多1个活跃Master。
  • Bootstrap指针：记录root tablet在哪台Tablet Server(整个BigTable数据的寻址起点)。
  • Tablet Server生命周期：启动时在Chubby指定目录下创建临时文件标志着上线了；Master监视该目录，文件消失认为下线了。
  • Schema存储：每张表的列族定义、权限ACL作为小文件放在Chubby。
  • 访问控制列表：客户端读ACL文件判断用户权限(文件需要从Chubby里获取)。

  如果Chubby不可用，那么Bigtable也将不可用。

3.5 执行过程

3.5.1 三大组件架构：

组件	作用	特点
Client Library	被链接到每个应用进程里	- 自己缓存tablet → 所在服务器的映射 - 读写直接连 Tablet Server，不经过 Master
Master	集群管理者	- 只负责元数据管理(分配 tablet、负载均衡、故障检测等) - 不承载数据流量 → 负载很轻
Tablet Server	工作者，被master调度	- 每台管理10–1000个tablet - 处理读写请求及tablet分裂 - 可动态上下线，弹性伸缩

3.5.2 数据流路径(关键设计)

• 客户端首次从Chubby拿到root tablet位置(bootstrap)。
• 随后自己递归(因为是多级索引)地从元数据表(METADATA 表)查出目标tablet所在服务器。
• 直接与对应的Tablet Server建立连接读写数据，因此大部分的读写请求不会经过Master，Master只做后台调度，不会成为性能瓶颈。

显然这是基于DFS的设计。

3.5.3 数据分片模型

• table → 多个tablet，每个tablet对应一段连续的行键范围(row range)。
• 初始：表只有1个tablet，表可能有多个。
• 自动分裂：当tablet大小达到100–200MB(默认阈值)，Tablet Server会就地拆成两半，并异步通知Master重新登记，分裂过程对客户端透明，无需停机。

Row提到过tablet的作用，选择100-200MB作为阈值是因为太小操作过于频繁，太大的话占用内存过大，并发效率降低，可以联想到DFS选择64MB为默认chunk，这样也不会多出很多碎片文件浪费存储空间。

3.5.4 弹性伸缩

• Tablet Server可动态加减
  • 上线：向Chubby注册临时文件，Master发现后把空闲tablet迁移过来(负载均衡)。
  • 下线/宕机：临时文件消失，Master将其tablet重新分配给其他服务器->集群容量随业务负载平滑扩缩。

这也是前面提到过的Chubby的核心作用之一。

3.6 总结

可以看出来BigTable非常依赖GFS这个底层框架，也可以看出来Google工程师的厉害之处。当然BigTable也还有很多东西没有整理，包括tablet位置与动态分区、SSTable底层结构等，但已经整理了其核心内容，剩下的细节设计部分就交给以后的我吧。。。