现代数据库及大数据管理—常见问题与技术归纳

1. 多结构化数据管理概述

内容：

数据的存储和管理机制
案例、现有的技术
典型系统原理及其优化
基本思路、基本方法
特殊类型数据处理

2. 多结构化数据存储与组织方法

2.1. memcached

memcached是一个高性能的分布式内存对象缓存系统，用于动态web应用以减轻数据库负载。

2.1.1. 掌握：数据分布式策略

Memcached的服务端并没有“分布式”功能，仅包括内存存储功能，实现非常简单。分布式完全由客户端程序库实现。

例如：向服务器添加键名为“ranme”的数据——

向memcached中添加“rname”键；
“rname”传给客户端程序库后，客户端算法根据“键”来决定保存数据的MemCached服务器节点；
客户端命令服务器节点保存“rname”及其值。

获取数据——
1.将要获取的键传递给函数库；
2.函数库通过与数据保存时相同的算法，根据“键”选择服务器；
3.发送get命令。
只要数据仍在服务节点缓存中就能获得保存的值。

Memcached通过将不同的键保存到不同的服务器上实现了分布式。服务器增多后键会分散，即使一台memcached服务器发生故障，也不影响其他缓存节点，系统依然能继续运行。

2.1.2. 重点掌握：一致性hash

Memcached的标准的分布式方法（对键的存储根据服务器台数的余数进行分散）：

求得键的整数哈希值；
除以服务器台数，根据其余数来选择服务器。
当选择的服务器无法连接时，rehash——将连接次数添加到键之后再次计算哈希值并尝试连接。

优点：方法简单，数据的分散性一般较好。
缺点：当添加或移除服务器时，缓存重组的代价大（所有键都需要重新计算所在服务器，然后重新分配）

改进的分布式方法——Consistent Hashing（一致性hash）：

求出服务器节点的哈希值，将其配置到0～232的圆上；
用同样的方法求出存储数据的键的哈希值并映射到圆上；
从数据映射到的位置开始顺时针查找，将数据保存到找到的第一台服务器上；
如果超过232仍然找不到服务器，就保存到第一台服务器上。

为什么可以解决缓存重组的问题？
添加一台服务器时，只有环上增加服务器位置的逆时针方向第一台服务器之间的键会受到影响，从而有效抑制了键的重新分布。

2.1.3. 掌握：虚拟节点

为什么要添加虚拟节点？
使用一般的hash函数，服务器的映射地点的分布可能出现不均匀的情况。

虚拟节点：
为每个物理节点（服务器）在圆环上分配100～200个点，从而抑制分布不均匀，最大限度地减小服务器增减时的缓存重新分布。

2.2. DynamoDB

DynamoDB是Amazon（亚马逊）的key-value模式的存储平台。

2.2.1. 对一致性hash的优化

主要缺点：虚拟节点位置和大小随机，如果有新节点加入，可能导致扫描所有节点上的所有数据对象，判断全部数据对象是否需要迁移，这种全局的扫描造成很大的开销。

优化：固定所有虚拟节点的大小和位置，只改变虚拟节点和节点的对应关系。

将整个地址空间平均分成Q个虚拟节点，每个物理节点（假设有S个）分配Q/S个虚拟节点。当有节点加入时，从现有节点每个拿出等量的虚拟节点分给新节点；当有节点离开时，将此节点的所有虚拟节点平均分配给余下的节点，保证系统中每个节点始终都有Q/S个虚拟节点。

固定所有虚拟节点的大小和位置的策略下，为实现结点加入和离开系统时影响较小，要求Q>>（远大于）S。
如果业务快速发展，使得主机不断增加，从而导致Q不再满足Q>>S
↓
该策略将不断退化。(实际应用中还有存储备份的因素，假设每个数据存储N个备份，则要满足Q>>S*N)。

2.3. 应该学习的：Gossip（谣言）协议

模拟人类中传播谣言的行为。传播谣言首先有种子节点，种子节点每秒随机向其他节点发送自己所拥有的节点列表以及需要传播的消息。
↓
任何新加入的节点在这种传播方式下很快地被全网所知道。

效果：设计时并没有刻意让信息一定要传递给所有的节点，但是随着时间的增长，在最终的某一时刻，全网会得到相同的信息。这个时刻可能仅仅存在于理论，永远不可达。

2.4. 重点掌握：时钟向量——Vector Clock

2.4.1. 基本知识点

逻辑时钟定义：
单机系统易于给发生的所有事件定义一个全局顺序，但是分布式系统没有全局时钟，很难给所有事件定义全局顺序。为此Lamport定义了逻辑时钟（Logical Clock），算法如下：

每个机器i本地维护一个逻辑时钟LCi
每个机器本地每发生一个事件设置LCi = LCi + 1，并把结果作为该事件的逻辑时钟。
当机器i给机器j发送消息m时，把LCi存在消息里。
当机器j收到消息m时，
LCj = max(LCj, message_timestamp)+1，结果值作为收到消息m这个事件的时间戳。

Vector Clock：
每个机器节点维护一个逻辑时钟的向量VC。
向量VC有如下属性：

VCi[i]是到目前为止节点i本地发生事件的个数；
VCi[k]是节点i知道的节点k发生的事件的个数(即节点i对节点k的认识)。

Vector Clock的更新算法：

节点i本地发生一个事件时，VCi[i]加1；
节点i给节点j发送消息m时，将整个VCi存在消息内；
节点j收到消息m时，VCj[k]=max(VCj[k],VCi[k])，同时VCj[j]加1。（所有分量）

时钟向量的作用
Vector Clock是一种维护因果关系(causality)的手段：
Vector Clock在节点间传递，从而给对方传递自己知道的其他节点的知识。
通过比较这些VC向量的大小，可以分析事件发生的顺序：
假如一个向量的所有分向量的值都小于或等于另一个向量 → 后者比前者更“新”；
否则存在冲突。

2.4.2. 集群节点数增加时钟向量开销会无限增大的问题

现实生活中，如果有很多的决策者，相当于有很多的客户端，整个向量时钟的长度就无限制增长了，这对于存储系统来说，需要想办法解决。
有两种优化：

1. 服务器向量
server的数量是可控的，可以不用client来标识向量空间，而是用server来标识向量空间——向量标签不再是客户端，而是server标识。

仍然存在问题：会因为网络传输延时而丢失数据。

2. 向量剪枝

所以为了解决向量时钟空间的无限增长问题，引入了向量时钟的剪枝。

Riak用四个参数来避免向量时钟空间的无限增长：

small_vclock
big_vclock
young_vclock
old_vclock

small_vclock和big_vclock参数标识向量时钟的长度，如果长度小于small_vclock就不会被剪枝掉，如果长度大于big_vclock就会被剪枝掉

young_vclock和old_vclock参数标识存储这个向量时钟时的时间戳，剪枝策略同理，大于old_vclock的才会被剪枝掉

这样只会丢掉一些向量时钟的信息，即数据更新过程的信息，但是不会丢掉实实在在的数据。只有当一种情况会有问题，就是一个客户端保持了一个很久之前的向量时钟，然后继承于这个向量时钟提交了一个数据，此时就会有冲突，因为服务器这边已经没有这个很久之前的向量时钟信息了，已经被剪枝掉了可能，所以客户端提交的此次数据，在服务端无法找到一个祖先，此时就会创建一个sibling。

所以这个剪枝的策略是一个权衡tradeoff，一方面是无限增长的向量时钟的空间，另一方面是偶尔的会有"false merge"，对，但肯定的是，不会悄无声息的丢数据。综上，为了防止向量时钟空间的无限增长，剪枝还是比用server标识向量时钟工作的更好。

2.5. 如何快速校验副本一致性：Merkle Tree

为了更快地检测副本之间的不一致性，并且减少传输的数据量和反熵过程中磁盘读取的次数，Dynamo采用 Merkle Tree技术。
Merkle tree：
每个叶子节点对应一个数据项，并记录其hash值；
每个非叶子节点记录其所有子节点的hash值。
Dynamo为每一个分片维护一个Merkle Tree，需要比较分片是否相同时，自根向下的比较两个Merkle Tree的对应节点，可以快速发现并定位差异所在。

Merkle Tree的主要优点是树的每个分支可以独立地检查，而不需要下载整个树或整个数据集。
此外，MerkleTree有助于减少为检查副本间不一致而传输的数据的大小。
例：如果两树的根哈希值相等，且树的叶节点值也相等，那么节点不需要同步。如果不相等，则意味着一些副本的值不同。在这种情况下，节点可以交换子节点的哈希值，该处理一直进行到树的叶子，此时主机可以识别出“不同步”的 key。

2.6. 重点掌握：布隆过滤器（BloomFilter）

布隆过滤器可以帮助我们判断某个数据是否一定不存在。

例如：你有10个Url，你完全可以创建一长度是100bit的数组，然后对url分别用5个不同的hash函数进行hash，得到5个hash后的值，这5个值尽可能的保证均匀分布在100个bit的范围内。然后把5个hash值对应的bit位都置为1，判断一个url是否已经存在时，一次看5个bit位是否为1就可以了，如果有任何一个不为1，那么说明这个url不存在。这里需要注意的是，如果对应的bit位值都为1，那么也不能肯定这个url一定存在，这个是BloomFilter的特点之一。

BloomFilter的核心思想有两点：

多个hash，增大随机性，减少hash碰撞的概率
扩大数组范围，使hash值均匀分布，进一步减少hash碰撞的概率。

BloomFilter的应用

黑名单
比如邮件黑名单过滤器，判断邮件地址是否在黑名单中
排序(仅限于BitSet)
仔细想想，其实BitSet在set(int value)的时候，“顺便”把value也给排序了。
网络爬虫
判断某个URL是否已经被爬取过
K-V系统快速判断某个key是否存在
典型的例子有Hbase，Hbase的每个Region中都包含一个BloomFilter，用于在查询时快速判断某个key在该region中是否存在，如果不存在，直接返回，节省掉后续的查询。

3. 多结构化数据管理存储与组织方法二

以某个典型系统为例，介绍其从数据存储结构从副本策略到集群存储再到分布式存储的发展脉络。

4. 谷三篇——大数据的开篇

4.1. 分布式文件系统GFS

4.1.1. 基本概念

Chunk：文件被分割成固定大小的Chunk。
每一个Chunk创建时，Master服务器给其分配一个64位全球唯一且不变的Chunk标识。

Namespace(名称空间)：GFS提供了一套类似传统文件系统API的接口函数（没有严格按照POSIX等标准API的形式实现），文件以分层目录的形式组织，用路径名来标识。
在逻辑上，GFS的名称空间就是一个全路径和元数据映射关系的查找表。

4.1.2. 要掌握的重点：解决一致性的问题

不是靠一致性算法，是靠写流程和租约、版本实现的。

4.1.2.1. 写数据——租约

GFS必须将对数据块的修改同步到每一个副本。

多个应用可能同时修改同一数据块
↓
并发应用下为使多个副本之间保持一致，需要为修改操作定义统一的时序。

问题：为减少master负担，系统采用了元数据缓存机制，client在获得并缓存副本位置后就不再和master交互，谁来定义时序？

需要有人代替master的角色
↓
选出一个master的代理来完成时序约定任务。
↓
租约（lease）机制：master建立租约，并将其授权给某个副本，称其为primary，由它来确定数据修改的顺序，其它副本照做。

例如：

客户机向Master节点询问哪个Chunk服务器持有当前租约，以及其它副本的位置。
如果没有Chunk持有租约，Master就选择其中一个副本建立一个租约。

这里有个问题，即如何选择租约副本？ 考虑可用性和负载均衡，即该chunk server是否可用，该chunk server当前的负载（带宽使用、磁盘利用率、cpu利用率等）情况以及性能（响应时间、吞吐量、系统容量等）。

Master将主Chunk的标识符以及其它副本（secondary副本/二级副本）的位置返回给客户机。客户机缓存这些数据以便后续操作。

此后只有在主Chunk不可用或主Chunk回复信息表明它已不再持有租约时，客户机才重新联系Master节点。

客户机把数据推送到所有的副本上（可按任意的顺序推送）。

Chunk服务器接收到数据并保存在它的内部LRU缓存中，直到数据被使用或过期交换出去。

当所有的副本都确认收到了数据后，客户机发送写请求到主Chunk服务器。

主Chunk为接收到的所有操作分配连续的序列号（这些操作可能来自不同的客户机，序列号保证了操作顺序执行），按序列号的顺序把操作应用到primary自己的本地状态中。
主Chunk把写请求传递到所有的二级副本。每个二级副本依照主Chunk分配的序列号以相同的顺序执行这些操作。
所有的二级副本回复主Chunk“它们已经完成了操作”。
主Chunk服务器回复客户机（任何副本产生的任何错误都会返回给客户机）。
出现错误时，写操作可能在主Chunk和一些二级副本执行成功。（如果主Chunk上操作失败，操作不会被分配序列号，也不会传递）客户端请求被确认为失败，被修改的region处于不一致状态。客户机代码通过重复执行失败的操作来处理这样的错误。在从头开始重复执行之前，客户机会先尝试几次步骤（3）~（7）。

4.1.2.2. 一致性保障机制-操作顺序、版本检测

经过了一系列成功的修改操作之后，GFS确保被修改的文件region是已定义的，并且包含最后一次修改操作写入的数据。
GFS确保上述已定义效果的措施包括：
（a）对Chunk的所有副本的修改操作顺序一致；
（b）使用Chunk版本号来检测副本是否因为它所在的Chunk服务器宕机而错过了修改操作而导致其失效。
失效副本处理：
（1）失效副本位置信息Master服务器将不再返回客户端；
（2）该副本不再进行任何修改操作；
（3）该副本会被垃圾收集系统尽快回收。
Chunk版本号增加：
Master节点和Chunk签订一个新租约时
↓
Master增加Chunk版本号，然后通知最新的副本。

Master和这些副本都把新的版本号记录在它们持久化存储的状态信息中。
该动作发生在任何客户机得到通知以前（也是对新版Chunk开始写之前）。
保护措施：（Master）
1）Master在通知客户机某个Chunk服务器持有租约、或者指示从哪个Chunk服务器进行克隆时，消息中都附带了Chunk的版本号。
2）Master在回复客户机的Chunk信息请求时会认为那些过期的Chunk不存在（即使过期副本还未被回收）。
3）Master例行垃圾回收时移除所有的过期失效副本。
**Chunk服务器失效：**Chunk副本可能因错失一些修改操作而过期失效。
↓
Master保存每个Chunk的版本号以区分当前副本和过期副本。
客户机或者Chunk服务器：在执行操作时都会验证版本号以确保总是访问当前版本的数据。
如果某个副本所在的Chunk服务器处于失效状态，则该副本版本号不会增加。
↓
该Chunk服务器重新启动并向Master报告其拥有的Chunk集合及版本号时， Master可检测出这些过期的Chunk。

如果Master看到一个比它记录的版本号更高的版本号，Master会认为它和Chunk服务器签订租约的操作失败了，此时会选择更高的版本号作为当前的版本号。

4.2. BigTable

提出多维度Map的概念而不是文件组织数据，通过行、列、列族、Tablet、SStable。

4.2.1. 重点：如何去做系统引导、如何寻址、缓存策略是什么？

4.2.1.1. 系统

Bigtable建立在其它几个Google基础构件上：
1）使用Google的分布式文件系统(GFS)存储日志文件和数据文件；
2）依赖高可用的、序列化的分布式锁服务组件Chubby。

Bigtable包括三类主要的组件：
1）链接到客户程序中的库
2）一个Master服务器
3）多个Tablet服务器

一个BigTable集群存储了很多表
每个表包含一个Tablet的集合
每个Tablet包含某个范围内的行的所有相关数据。
（ → SSTable → 数据块）

初始状态下，一个表只有一个Tablet，随着表中数据的增长，它被自动分割成多个Tablet。
缺省情况下，每个Tablet的尺寸大约是100MB到200MB。

4.2.1.2. 寻址

客户端寻址查找Tablet Server 的位置信息
Tablet的位置信息：使用一个三层的、类似Ｂ+树的结构存储。

Chubby file： 一个存储在Chubby中的文件，包含Root Tablet的位置信息。
第一层：Root Tablet，包含了一个特殊的METADATA表，记录METADATA里所有的Tablet的位置信息。
第二层：METADATA表，其中的每个Tablet包含了一个用户Tablet的集合。
第三层：用户表tablet。
Root Tablet：
1）实际上是METADATA表的第一个Tablet；
2）永远不会被分割，从而保证Tablet的位置信息存储结构不会超过三层。
METADATA：表中，每个Tablet的位置信息都存放在一个行关键字下
行关键字=Tablet所在的表的标识符
⊕Tablet的最后一行编码
METADATA表还存储次级信息(secondary information)，主要包括每个Tablet的事件日志（例如一个服务器何时开始为该Tablet提供服务）。

数据块的定位

BigTable内部存储数据的文件：Google SSTable。
SSTable是一个持久化的、排序的、不可更改的Map结构（key-value映射），key和value的值都是任意的字节串。

从内部看，SSTable是一系列的数据块，通常每个块的大小是64KB（可以配置）。

数据块的定位：使用块索引（通常存储在SSTable的末尾）定位数据块。
块索引缓存：在打开SSTable时块索引被加载到内存。
也可以选择把整个SSTable都放在内存中。

每次查找都可以通过一次磁盘搜索完成：
1）使用二分查找法在内存中的块索引里找到数据块的位置；
2）从硬盘读取相应的数据块。

4.2.1.3. 缓存策略分为客户端缓存、tablet服务器缓存

客户程序使用的库：缓存Tablet的位置信息，对它的操作不必访问GFS文件系统。

预取Tablet地址：每次需要从METADATA表中读取一个Tablet的元数据时，都会多读取几个Tablet的元数据。
↓
进一步的减少访问的开销。
客户端可能未缓存某个Tablet的地址信息（例如缓存为空），或发现缓存的地址信息不正确（过期）。
↓
客户程序在树状的存储结构中递归的查询Tablet位置信息。

Tablet服务器缓存
使用二级缓存策略：

第一级缓存——扫描缓存，主要缓存Tablet服务器通过SSTable接口获取的Key-Value对。

第二级缓存——Block缓存，缓存从GFS读取的SSTable的Block。

客户程序可以对特定局部性群组的SSTable指定Bloom过滤器缓存来减少硬盘访问的次数：
使用Bloom过滤器查询一个SSTable是否包含了特定行和列的数据(隐式的减少当应用程序访问不存在的行或列时对硬盘的访问）。

4.2.2. 分布式锁——Chubby

4.2.2.1. 较难的：Paxos

要求能描述proposer和acceptor各自的抽象算法。

基本原理：
法定集合：将一个超过半数的集合称之为法定集合，比如数字1、2、3、4、5，共5个元素，{1,2,3}有三个元素就是法定集合。
法定集合性质：任意两个法定集合，必定存在一个公共的成员。
该性质是Paxos有效的基本保障。（如果要回答paxos基于什么原理，法定集合的性质就是）

相关概念
instance(实例)：一次Paxos算法执行。
proposal（议案）：经发起而未经批准的提案。
value（决议）：被最终批准通过的议案中的value称为决议。
Proposer（提案者）：提出议案
Acceptor（批准者）：审批议案
Learner（学习者）：学习决议
基本约束条件：
1）决议（value）只有在被proposer提出后才能批准；
2）在一次Paxos算法的执行实例中，只批准一个Value；
3）learners只能获得被批准的Value。

抽象算法
proposer：
询问：询问法定集合进程的自身v值;
预提议：发送包含自身proposer-id的预提案给法定集合接受者;
提议：发送的预提案得到一个法定集合接受者的回复后，如果询问的结果是法定集合的接受者均未批准给v赋予值，那么提议者拥有自由赋值的权利，否则，提议者从中选择一个值赋予给v。假定自由赋予或者选择的值为c，发送包含c和proposer-id的提案给接受者。

acceptor：
处理询问：回复自身v的值
处理预提案：如果收到的预提案ppq.proposer-id>自身a-proposer-id，则更新自身a-proposer-id=ppq.proposer-id，并接受该预提案，否则拒绝这个预提案。
处理提案：如果收到的提案的pq.proposer-id>=自身 a-proposer-id，那么更新a-proposer-id=pq.proposer-id，接受该提案，记录v值为提案中的值，即v=pq.c。
（无论是提案还是预提案，接受者只接受proposer-id比它自身的a-proposer-id更大的消息。）

4.3. MapReduce

现实世界中的一些算法，用mapreduce描述出来。数学描述->mapreduce 编程模型->优化
重点问题：后代邻居、 pagerank

4.3.1. PageRank

PageRank，又称网页排名、谷歌左侧排名，是一种由搜索引擎根据网页之间相互的超链接计算的技术，而作为网页排名的要素之一，以Google公司创办人拉里·佩奇（Larry Page）之姓来命名。Google用它来体现网页的相关性和重要性，在搜索引擎优化操作中是经常被用来评估网页优化的成效因素之一。

假设一个由4个网页组成的群体：A，B，C和D。如果所有页面都只链接至A，那么A的PR（PageRank）值将是B，C及D的Pagerank总和。
PR(A) = PR(B) + PR© + PR(D)
重新假设B链接到A和C，C只链接到A，并且D链接到全部其他的3个页面。一个页面总共只有一票。所以B给A和C每个页面半票。以同样的逻辑，D投出的票只有三分之一算到了A的PageRank上。
PR(A) = PR(B)/2 + PR©/1 + PR(D)/3

对于一个页面A，那么它的PR值为：
[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-4akClywR-1574247698456)(_v_images/20191104140618113_8547.png =424x)]
PR(A) 是页面A的PR值
PR(Ti)是页面Ti的PR值，在这里，页面Ti是指向A的所有页面中的某个页面
C(Ti)是页面Ti的出度，也就是Ti指向其他页面的边的个数
d 为阻尼系数，其意义是，在任意时刻，用户到达某页面后并继续向后浏览的概率，该数值是根据上网者使用浏览器书签的平均频率估算而得，通常d=0.85

代码示例：

Main

1: Job job = new Job();
2: job.AddMap(Map Rank, 1);
3: job.AddReduce(Reduce Rank, 1);
4: job.AddMap(Map Aggregate, 2);
5: job.AddReduce(Reduce Aggregate, 2);
6: job.SetDistanceMeasure(ResultDistance);
7: job.SetFixedPointThreshold(0.1);
8: job.SetMaxNumOfIterations(10);
9: job.AddInvariantTable(#1);//L表
10: job.SetInput(IterationInput);
11: job.SetReduceInputCache(true);
12: job.SetReduceOutputCache(true);
13: job.Submit();

IterationInput //首次迭代输入L ∪ R0 ，后续输入Riteration−1

Input: int iteration
1: if iteration==1 then
2: return L ∪  R0;
3: else
4: return Riteration−1
5: end if

ResultDistance

Input: Key out key, Set vi−1, Set vi
1: return |vi.get(0)−vi−1.get(0)|;

Map_Rank

Input: Key k, Value v, int iteration
1: if v from L then
2: Output(v.url src, v.url dest, #1);
3: else
4: Output(v.url, v.rank, #2);
5: end if
//发送L表（#1）和Ri表（#2）

Reduce_Rank

//输入L表（#1）和Ri表（#2）
Input: Key key, Set values, Set invariantValues, int iteration
1: for url_dest in invariantValues do
2:    Output(url_dest, values.get(0)/invariantValues.size());
3: end for
//输出（url_dest, rank/count(url_dest)）
//同一个url下的url_dest, 每个（url, url_dest）输出一次rank

Map_Aggregate

//输入（url_dest, rank/count(url_dest)）
//每个（url, url_dest）对应一个输入（url_dest, rank）
Input: Key k, Value v, int iteration
1: Output(k, v.rank);

Reduce_Aggregate

Input: Key key, Set values, int iteration
1: Output(key, AggregateRank(values));
//输出（url_dest, sum(new_rank)）

4.3.2. 后代、邻居查询

F为社交网络关系表，寻找某人经过“两跳”之后的朋友。

** 代码示例**
Main

1: Job job = new Job();
2: job.AddMap(Map Join, 1);
3: job.AddReduce(Reduce Join, 1);
4: job.AddMap(Map Distinct, 2);
5: job.AddReduce(Reduce Distinct, 2);
6: job.SetDistanceMeasure(ResultDistance);
7: job.SetFixedPointThreshold(1); // 小于1时（即⊿Si个数为0）停止迭代
8: job.SetMaxNumOfIterations(2);
9: job.SetInput(IterationInput);
10: job.AddStepInput(StepInput);
11: job.AddInvariantTable(#1);
12: job.SetReducerInputCache(true);
13: job.Submit();

IterationInput

//初始输入为F∪⊿S0,后期为⊿Si-1。
Input: int iteration
1: if iteration==1 then
2:    return F ∪⊿S0;
3: else
4:    return ⊿Siteration−1
5: end if

Map Join

//初始输入为F∪⊿S0,后期为⊿Si-1。
Input: Key k, Value v, int iteration
1: if v from F then
2:     Output(v.name1, v.name2, #1);
3: else
4:     Output(v.name2, v.name1, #2);
//后期输出key为邻居姓名，值为本人
5: end if

Reduce Join

Input: Key key, Set values, Set invariantValues,
int iteration
1: Output(Product(values, invariantValues)); // Product:笛卡尔积运算
//输出同一个name1（/2）下的姓名及其间接关联的邻居姓名（发现的下一跳的邻居）
//通过相同的name2的reduce执行实现join，输出name1，F的name2

StepInput

//迭代中第二次Map-Reduce作业的输入会额外增加∪0≦j ≦(iteration−1) ⊿ Sj
Input: int step, int iteration
1: if step==2 then
2: return ∪0≦j ≦(iteration−1) ⊿ Sj
3: end if

Map Distinct

//输入为新邻居以及∪0≦j ≦(iteration−1) ⊿ Sj
Input: Key k, Value v, int iteration
1: Output(v.name1, v.name2, iteration);
//输出为∪0≦j ≦(iteration−1) ⊿ Sj中的两个姓名及其迭代下标。

Reduce Distinct

Input: Key key, Set values, int iteration
1: for name in values do
2:    if (name.iteration < iteration) then
3:       set_old.add(name);
4:    else set_new.add(name);
5: end for
6: Output(Product(key, Distinct(set_new – set_old)));
//同一个key（name）下的第i跳的邻居

ResultDistance

Input: Key out key, Set vi−1, Set vi
1: return vi.size();