分布式系统设计自学整理

下周面试，希望能用一周时间按照自己的逻辑整理一下之前所学的分布式系统知识！


补充内容 (2018-11-9 09:57):
计划目录：
1. GFS
2. BigTable(SSTable)
3. Cassandra
4. MapReduce
5. Spark
6. Zookeeper
7. Kafka

1. GFS，先从最底层数据库结构开始了解，很多内容没有完全消化，这是我一次读paper的总结，希望以后慢慢了解更多，错误的地方希望大家指正！

https://pdos.csail.mit.edu/6.824/papers/gfs.pdf

GFS(Google file system): a distributed file system, 用来存储数据，作为BigTable（SortedString Table的底层）， 也是HDFS的真身。把文件放在很多个disks上，能够满足大量读写需求。

GFS的目的：

1. Fault-tolerance: 假设有1000个机器，每个机器每年会坏一次，那么每天就有大概3个机器会挂。如何确保数据安全？

2. Performance： 大量的同时读写，如何处理I/O

3. Consistency：如果有多个人同时读写，如何确保最后读到的是最新的数据

4. Efficiency：如何最优使用网络bandwidth来传输数据

在GFS之前的文件系统：

1. 每个文件分成blocks （4096byte），每个文件有自己的Metadata，记录文件名/时间/size/和blocks的diskoffset

2. Master-slave 系统：数据都存放在master上，slave是备份，master挂了，找一个slave顶上去

3. 缺点：随机访问速度慢，多进程读写需要锁进程，数据存放在一个master严重依赖badnwidth

GFS的结构：

1. client，比如SSTable，read/writefile to GFS

2. Master，这里的master跟一般文件系统的不一样，不存放数据，memory中存放

3. Chunk servers，存放数据的metadata和一个hashtable[filename][chunk_index] = chunk_server

4. chunks，和一般的4096 block不同， GFS的数据每一块比较大，default用的是64M，这样metadata就比较小了。建立在小数据碎片很小的基础上。

5. Megadata, 64Byte 每个chunk

6. Replica： 每个数据存3次，在不同的chunkserver上，以防止chunk server 挂了

Read：读数据：

1. Client 发送 filename 到master

2. 在master 找到这个file的metadata， 然后找到相应的chunk list，最后在hashmap里找相应的chunk serverlist，返回给client，这些数据会在client的cache里

3. Client读chunk server list， 对于每个chunk，对应最多3个server，根据IP地址选择最近的server来读取数据

*4. 实际上会有一个 version的信息，在每个数据里和chunk server list，如果在server数据里的version和chunk server 不同就重新连接master

*5. Client 会吧server list 放进cache里，这样就不需要每次都问master了

Write: 写数据：（这里的写指的是modify）

1. Client发 filename到master

2. master恢复chunk server list，这里跟读一样，但是chunkservers分成两种，一个叫primaryreplica，一种叫secondary replica

3. 把数据写到三个replica的cache里（！！不是硬盘，是LRU cache）

4. Client 发写请求给primary replica，primary把最新的文件写到硬盘

5. Primary 把写请求 forward给两个secondary， secondary  写data

6. 写完之后secondary告诉primary写好了

7. primary 回复client， 如果出错就重复3-7

*和传统方法不一样的是，GFS写数据提供recordappend方法，用来保证多client同时写

系统优化：

1. 一般不需要第二个master， 如果挂了去重启就行，但是master会存一个log到硬盘，重启之后从log恢复state

2. 如何判断chunk server是不是挂了：heartbeat message

3. 如何判断一个chunk server上某个block挂了： Achunk is broken up into 64 KB blocks. Each has a corresponding 32 bit checksum.

MapReduce

这里主要描述的是MapReduce 系统的构建，因为我简历里面写了MapReduce，之前在网上也看了挺多比如如何用mapreduce解决排序这样的问题，后来一次面试的时候挂在了讨论如何设计MapReduce系统中的master，然后就意识到了，还是要了解系统构建的，毕竟以后的工作可能是自己给MapReduce写client方或者Server方的API这样的工作，而不是用MapReduce跑数据。

在这之前先看一下我们（或者面试官）可能关注的点：

1.Implement： 如何用多线程和Remote procedure call 来实现功能

2.Performance：如何衡量并提高系统的performance，比如scaling，问题的关键是要找出系统的bottleneck在哪里

3.Fault Tolerance： 如果某个机器挂了怎么办, 如果数据挂了怎么办

4.Consistency：如何保证replica都是一样的，如何保证log和内存和disk是一样的。很难保证100%的consistency，那样performance会很差，但是可以满足weak consistency。

然后来看MapReduce：

MapReduce想要做的事情，就是用distribute的思想来处理大数据，举个例子：

input: 一个文件，100行，文件可以是NoSQL的key-value形式

key  value

a      1

b      2

c      3

a      31

…     …

output：一个按照key排好序的文件

key  value

a     1

a     2

a     31

…    …

如果我的计算机内存只能读取10行，但是有无限个计算机，怎么做？

最简单的distributed思想是，先把文件分成10个，每个文件10行，然后用10个计算机排序（每个计算机时间是O(10log10)）然后输出10个排序好的文件，最后用第11个计算机，来一个10路归并（时间 O(100log(10))），同时输出成最后的output。所以在这个算法里，bottleneck是最后那个10路归并。

MapReduce的处理方法是这样的：

首先分成10个文件，然后输出的不是10个文件，而是260个，比如第一个计算机输出的26个文件是xxx_1_a.txt,xxx_1_b.txt … 分别对应key开头字母为a, b,c ..

最后merge的时候，用26台计算机，比如第一个台专门处理key开头字母为a的那些file：xxx_1_a.txt,xxx_2_a.txt …这样以来bottleneck就被解决了

MapReduce工作流

1.把数据分成M个小文件(每块大概16M-64M),然后把map-reduce的程序copy给workers和 master

2.Master管理workers来执行map，reduce。以map为例子，master有一个map workers pool，还有一个map taskspool，workers pool里找到worker，来执行map task，worker完成任务之后回收到pool里等待被分配下一个task

3.Map worker 收到一个文件还有map function， 开始执行mapfunction，比如sort，然后把结果存在bufferedmemory中，每隔一段时间，再按照reduce的key 分类规则写入到local disk（比如abcd字母开头，这是一种规则）。完成之后，把这些文件的地址返回给master

4.Master确认所有map tasks pool中任务都被执行了之后，建立reducetasks pool，从reduce workers pool中分配机器来完成reduce任务

5.Reduce worker 通过RPC收到file list，还有 reduce function， 根据intermeidatekey排序这些files （外排序），并且写入到最终文件

6.当所有的Reduce tasks被完成之后，Master向user 汇报完成任务

然后我们按照之前说过的点一个一个看：

Implement： MapReduce系统实现

Master:

数据结构：

-Mutex 进程锁

-string workers[]: workers RPC address

-string files[]: input files

-int nReduce: reduce的数量

-int stats[]

functions:

+Register: worker call  register来汇报自己状态，同时存储这个worker到worker pool

+newMaster: initial 一个 master

+killWorker

+run:从worker pool里选择worker来执行task

Worker：

数据结构：

+map

+reduce

-name

-nTasks

-nConcurrent

functions:

+run

+register

+shutdown

Performance：

提高performance最重要的是找到系统的bottleneck，对于mapreduce，bottleneck就是 RPC的网速带宽。intermediatedata是存放在local disk，但是input 和 output是在GFS上进行的。这里的解决办法是，因为GFS上每个file都有3份replica，所以Master会根据3份replica的地址选择map worker，理想情况下，map worker处理的这个input正好在这个disk上，相当于从local读取数据

Faulttolerance：

Master挂了：换个master整个重新做

Worker挂了： master每隔一段时间就ping一下works，如果一段时间没回应就是挂了

Mapworker：可能是replica出了问题，也可能是worker出了问题，所以换一个replica，同时换一个worker重新执行

Reduceworker：如果任务已经finish了，重启就行了；如果任务没有finish，那么把挂掉的worker设置成idle，重新选一个worker执行这个任务

Consistency：

Itis a deterministic system. So the consistency should be good.

如果两个worker同时做一个map，会生成两个file，reduce只读其中一个

如果两个worker同时做一个reduce，在GFS中只会有一个可见

总结：

MapReduce适用于离线大数据分析

不适合iterative任务，因为每次要写入读出到GFS

不适合小数据，以及低延迟的系统

BigTable: a NoSQL database (Column based) based on GFS

补充：大数据中常见的两种NoSQL：1）key/ value based like Redis or MemcacheDB 和 2）column based， BigTable

SSTable的数据结构:

1. Rows (key) : 最大64KB 的字符串，在SSTable中，是按照Row key 排序好的，顾名思义，sorted string table

2. Columnkeys: 一个set，可以支持range query

3. value：包括两部分，第一部分是value， 第二部分是时间timestamp，是一个64bit的整

SSTable的文件格式：原则是不能修改，只能append，根据timestamp来决定哪个是最新的value

1. 分成datablock， 每个block 64

2. 在SSTable最后有一个 block inde

3.查询SSTable先 load index然后二分查询 （不能放进内存用hashtable查询！）

Bigtable 系统:

1.Master server: 负责测试tablet server是不是在工作， 处理GFS的sharding等，在Bigtable中，Client是不会和master交流的，要和Lock交流

2. Lock server: 用Chubby或者zookepper实现，完成多进程的锁，用metadata查找key所在的server

3. Tablet server: 处理读写操作

写操作：

1.Client ask lock

2.Lock return a tablet server and lock it

3.Go to tablet server, 1) write to 1) commit log (write ahead log) in disk and 2)memTable in memory. If memory down, recovery from log

4. Minorcompaction: when memory hit the threshold, frozen this one and write to GFS asa SSTable

5.Major compaction: Merge SSTable, with same row key, use the most recent record.

6.After all, client ask lock to unlock the tablet server and update the metadata

读操作：

1. Clientask lock

2.Check metadata, return the tablet server and lock it

3.Go to tablet server, first check memTable

4.If not in memTable , check tablets one-by-one

5.For each tablet, check the index firstly. So the worst time is O(mlogk), m isthe number of tablets, k is the length of each tablets.

6.Retern value and unlock the server

其他重要的scale优化：

1. readcache: 在tabletserver上，1）scan cache 保存已经read的key/value，优化重复read； 2）block cache，存在GFS上SSTBlock，优化读取附近的key（不是很懂）

2.Bloom filter：对每个SSTable加入一个bloom filter (多个hash函数来确定的一个bit filter)，如果key没有通过SSTable的bloom filter，就不用read SSTable的index了，省去了二分的时间

amcw7777 发表于 2018-11-9 09:43
1. GFS，先从最底层数据库结构开始了解，很多内容没有完全消化，这是我一次读paper的总结，希望以后慢慢了 ...

分享一个 levelDB (SSTable) 资料：leveldb-handbook.readthedocs.io/zh/latest/

一般不需要第二个master， 如果挂了去重启就行，但是master会存一个log到硬盘，重启之后从log恢复state

这样不是就有downtime了么？另外，NameNode在现实中承受的大量QPS，很容易挂掉。现在大一点的文件系统Metadata都是分布式存储的。

Euler57721 发表于 2018-11-10 01:21
这样不是就有downtime了么？另外，NameNode在现实中承受的大量QPS，很容易挂掉。现在大一点的文件系统Met ...

谢谢大大的回复！我想了一下你的问题：
1. Master建立自己的replica，同时setup shadow master。Shadow master平时设置成read-only，可以在master当机的时间load replica，并且代替master工作。
2. 如果64M的data，有64B的megadata，那么1PB的data，megadata也就1G，大小足够放在内存了，假设放在了memcached里，100k的QPS，应该足够多的hadoop来map了？是不是真正工业界的service scale比我想象的大很多啊。。
3. 请教一下大大，我过几天要去databricks面试，因为我是new grad不大了解工业届，你觉得我应该跟面试官聊啥？谢谢！

amcw7777 发表于 2018-11-10 01:46
谢谢大大的回复！我想了一下你的问题：
1. Master建立自己的replica，同时setup shadow master。Shadow  ...

回答你的问题。
2. 现实生活中data不总是占满64M，会有大量几k几十k的文件碎片，1PB的metadata至少在几十G。大一点公司的storage QPS都是用million计的。
3. 面Databricks你应该研究Spark和Flink，他们又不做Storage

补充内容 (2018-11-10 02:19):
另外，作为面试官我肯定会考的问题：
1. 怎么保证Master和Shadow master一致性？写一半挂了怎么半？

谢谢～
加入QPS是10M， 每个内存的QPS是100k，所以需要100个server来负责matadata的处理？那样的话一致性就很难确保了呀，是有什么特别的方法吗？
我的计划是一会把p2p的NoSQL整理一下，然后就走一下MapReduce -> Spark，我觉得对数据库理解深刻一点对后面系统的bottleneck可能理解更深刻，加上这几篇paper确实写的挺精彩的！

Euler57721 发表于 2018-11-10 02:15
回答你的问题。
2. 现实生活中data不总是占满64M，会有大量几k几十k的文件碎片，1PB的metadata至少在几 ...

我觉得，可以为master 建立一个private database来写入 log，如果shadow有lag的话，可以从log中读取。
问题：如果有大量的写操作，那么master写入log的I/O会成为一个bottleneck
我记得Raft可以解决这个问题，但是具体原理我不是很清楚。
刚才我回复了自己，粘贴一下：
加入QPS是10M， 每个内存的QPS是100k，所以需要100个server来负责matadata的处理？那样的话一致性就很难确保了呀，是有什么特别的方法吗？
我的计划是一会把p2p的NoSQL整理一下，然后就走一下MapReduce -> Spark，我觉得对数据库理解深刻一点对后面系统的bottleneck可能理解更深刻，加上这几篇paper确实写的挺精彩的！

真的很谢谢你！ 希望能指出更多我理解不到位的问题！

Cassandra，简单看一下，主要总结一下consistanthashing，今天重点是MapReduce，如果以后有需求再revisit

特点：NoSQL，peer-to-peer，column key

主要技术：consistant hashing

特点：容易扩展（consistant hashing），单值读写快，一致性高

Consistent hashing:

1. 加入有一个环，等分成2^64份

2. 现在加入一个机器A，在环上随机3个位置设置成A的三个virtual nodes

3. 来一个key，hash成0-2^64的一个值，从这个值顺时针找到第一个virtual node，就把这组（key，value）存放在这个node上

4. 加入新机器，比如已经有A,B加入C，还是随机生成3个C的virtual nodes。然后把C到A之前的data从Btransfer到C

5. Repica： 不仅仅存在key后面第一个virtual node上，而是存在三个node上