Lab 1 实验 MapReduce

👂 若月亮没来 (若是月亮还没来)（若是月亮还没来） - 王宇宙Leto/乔浚丞 - 单曲 - 网易云音乐

🌼参考代码

🐙解析

🐟mrsequential.go

🐟mrapps/wc.go

📕实验--准备

🎂概念

🐋思路梳理

🦖注意要点

🐆初始代码--研读

main/mrcoordinator.go

main/mrworker.go

mr/coordinator.go

mr/worker.go

mr/rpc.go

🦈实验--开始

🐋伪代码

mr/coordinator.go

mr/worker.go

mr/rpc.go

🐎结果

🌼参考代码

🐙解析

实验原文要求仔细研读两份代码，并在作业过程中大胆借鉴

wc.go

MapReduce 的插件，实现了 Map 和 Reduce 两个函数
Map 函数接收输入文本的内容，分割成单词，并为每个单词生成一个键值对（键是单词，值是1）
Reduce 函数接收 Map 生成的所有键值对，统计每个单词出现次数，并返回这个次数

mrsequential.go

MapReduce 的主体，协调 Map 和 Reduce 任务的执行
检查命令行参数
（
os.Args[0] // 可执行文件
os.Args[1] // "wc.so"（插件）
os.Args[2] // "pg1.txt"（输入文件）
os.Args[3] // "pg2.txt"
）
加载 wc.go，执行其中的 Map 和 Reduce 函数
读取输入文件内容，对每个文件调用 Map 函数
Map 函数的输出按键排序（输出单词）
排序的键值对进行 Reduce（归约）
Reduce 函数的输出写入输出文件（统计次数）

🐟mrsequential.go

代码是单线程的，输入 --> Map --> sort --> Reduce --> 输出

package mainimport ("fmt""6.824/mr"      // 引入MapReduce相关的数据结构和接口"plugin"       // 用于动态加载插件"os"           // 用于操作系统相关的功能，如命令行参数"log"          // 用于日志记录"io/ioutil"    // 用于I/O操作，如读取文件"sort"         // 用于排序
)// ByKey 是一个用于按键排序的切片类型
type ByKey []mr.KeyValue// Len 实现了 sort.Interface 接口的 Len 方法
func (a ByKey) Len() int { return len(a) }// Swap 实现了 sort.Interface 接口的 Swap 方法
func (a ByKey) Swap(i, j int) { a[i], a[j] = a[j], a[i] }// Less 实现了 sort.Interface 接口的 Less 方法
func (a ByKey) Less(i, j int) bool { return a[i].Key < a[j].Key }func main() {// 检查命令行参数数量if len(os.Args) < 3 {fmt.Fprintf(os.Stderr, "Usage: mrsequential xxx.so inputfiles...\n")os.Exit(1)}// 加载插件中的 Map 和 Reduce 函数mapf, reducef := loadPlugin(os.Args[1])// 用于存储Map阶段的中间输出intermediate := []mr.KeyValue{}// 遍历所有输入文件for _, filename := range os.Args[2:] {// 打开文件file, err := os.Open(filename)if err != nil {log.Fatalf("cannot open %v", filename)}// 读取文件全部内容content, err := ioutil.ReadAll(file)if err != nil {log.Fatalf("cannot read %v", filename)}file.Close()// 调用 Map 函数处理文件内容kv := mapf(filename, string(content))// 将Map结果添加到中间输出intermediate = append(intermediate, kv...)}// 对中间输出按键排序sort.Sort(ByKey(intermediate))// 创建输出文件oname := "mr-out-0"ofile, _ := os.Create(oname)// 调用 Reduce 函数并写入输出文件i := 0for i < len(intermediate) {j := i + 1for j < len(intermediate) && intermediate[j].Key == intermediate[i].Key {j++}// 收集相同键的所有值values := []string{}for k := i; k < j; k++ {values = append(values, intermediate[k].Value)}// 调用 Reduce 函数output := reducef(intermediate[i].Key, values)// 按格式写入输出文件fmt.Fprintf(ofile, "%v %v\n", intermediate[i].Key, output)i = j}// 关闭输出文件ofile.Close()
}// loadPlugin 从插件文件中加载 Map 和 Reduce 函数
func loadPlugin(filename string) (func(string, string) []mr.KeyValue, func(string, []string) string) {// 打开插件p, err := plugin.Open(filename)if err != nil {log.Fatalf("cannot load plugin %v", filename)}// 查找 Map 函数xmapf, err := p.Lookup("Map")if err != nil {log.Fatalf("cannot find Map in %v", filename)}mapf := xmapf.(func(string, string) []mr.KeyValue)// 查找 Reduce 函数xreducef, err := p.Lookup("Reduce")if err != nil {log.Fatalf("cannot find Reduce in %v", filename)}reducef := xreducef.(func(string, []string) string)// 返回 Map 和 Reduce 函数return mapf, reducef
}

🐟mrapps/wc.go

Map 函数返回键值对切片，Reduce 函数将单词出现次数转化为字符串后返回

// 定义包名为 main，这是一个插件，可以被 MapReduce 框架动态加载。
package main// 导入 MapReduce 框架的包，用于实现 Map 和 Reduce 函数。
import "6.824/mr"
// 导入 unicode 包，用于判断字符是否为字母。
import "unicode"
// 导入 strings 包，用于字符串操作。
import "strings"
// 导入 strconv 包，用于字符串和基本数据类型之间的转换。
import "strconv"// Map 函数是 MapReduce 框架中的第一个阶段，它将对输入文件的每一行调用一次。
// 参数 filename 是输入文件的名称，contents 是文件的全部内容。
// 这个函数返回一系列键值对，其中键是单词，值是 "1"。
func Map(filename string, contents string) []mr.KeyValue {// FieldsFunc 函数将根据 ff 函数来分割字符串。// ff 函数是一个过滤函数，它返回 true 如果字符不是字母。ff := func(r rune) bool { return !unicode.IsLetter(r) }// 使用 FieldsFunc 函数根据 ff 函数分割 contents 字符串，得到单词数组。words := strings.FieldsFunc(contents, ff)// 初始化一个空的键值对切片，用于存储 Map 函数的输出。kv := []mr.KeyValue{}for _, w := range words {// 对于每个单词 w，创建一个键值对，键是单词本身，值是 "1"。kv = append(kv, mr.KeyValue{w, "1"})}// 返回包含所有单词和计数的键值对切片。return kv
}// Reduce 函数是 MapReduce 框架中的第二个阶段，它对每个唯一的键调用一次。
// 参数 key 是键，values 是所有映射任务为该键生成的值的列表。
// 这个函数返回一个字符串，表示键出现的次数。
func Reduce(key string, values []string) string {// 使用 strconv.Itoa 函数将 values 切片的长度（即 key 出现的次数）转换为字符串。return strconv.Itoa(len(values))
}

📕实验--准备

🎂概念

所谓“单机”，整个项目部署在一台机器上

所谓“集群”，集群中的每一个节点就是一个单机，每个单机运行同一个的项目，通过设置“调度者”，用户请求先发送到“调度者”，再由“调度者”根据所有节点的负载情况，分配任务，即负载均衡

（从单机到集群，代码无需修改，只需多部署几台服务器）

所谓 “分布式”，类似流水线（只是将串行改成了并行），每条线负责不同的功能，最终将一个个小功能，整合成一个项目

（也就是将原本部署在单机上的系统，拆分成一个个子系统，每个子系统都是独立的）

这些子系统存在依赖关系，在网络中通过 rpc(remote procedure call) 通信

🐋思路梳理

wc.go 是一个实现了 Map 和 Reduce 函数的插件

而 mrsequential.go 是 MapReduce 的顺序实现

（可以理解为“单机”实现，一台机器，单个进程，顺序执行）

我们要做的就是，将 mrsequential.go 拆分成 5 个文件，实现 MapReduce（词频统计）的分布式部署 / 并行执行

main（程序入口）

main/mrcoordinator.go 协调者初始化
main/mrworker.go 工作者初始化

两个 main 文件不用修改，我们只需完成以下 3 个 mr/.... 文件即可

mr（具体实现）

mr/coordinator.go 实现协调者（监视 worker，分配任务，处理失败，重新分配）
mr/worker.go 实现工作者（请求任务，执行 Map，执行 Reduce，写入中间结果，写入最终结果）
mr/rpc.go 协调者与工作者间的远程调用 （定义了通信接口和数据结构）

🦖注意要点

修改 mr/ 下任何文件后，需要重新构建插件 wc.go，确保插件不依赖旧版本
go build -race -buildmode=plugin ../mrapps/wc.go
修改 mr/worker.go 中的 Worker() 函数，通过 RPC 请求 coordinator 分配任务
中间文件命名 mr-X-Y（X 为 Map 任务编号，Y 为 Reduce 任务编号）
使用 Go 的 encoding/json 包写入和读取 JSON 文件
enc := json.NewEncoder(file) // json.Encoder实例，编码为 json 格式
for _, kv := ... {err := enc.Encode(&kv)
dec := json.NewDecoder(file) // json.Decoder 实例，解码为 json 格式
for {var kv KeyValueif err := dec.Decode(&kv); err != nil {break}kva = append(kva, kv)
}
使用 mrapps/crash.go 插件测试崩溃恢复
go build -race -buildmode=plugin crash.go // 编译插件文件
go run -race mrcoordinator.go pg-*.txt // 根据输入文件，启动 MapReduce 作业
go run -race mrworker.go crash.so // 运行 worker 进程，使用插件故意崩溃
为防止崩溃时部分写入，用 ioutil.TempFile 创建临时文件，os.Rename 原子地重命名

🐆初始代码--研读

初始代码可以先抄一遍，理解一下，捋清楚思路后，再开始做

main/mrcoordinator.go

创建协调者，通过命令行参数，传递输入文件给工作者，并在作业完成后退出程序

// 程序入口点
package main// 引入 MapReduce 包，包含协调者和工作者的实现
import "6.824/mr"
// time 包，暂停时间
import "time"
// 引入 os 包，读取命令行参数。
import "os"
// 格式化输入输出
import "fmt"func main() {// 访问命令行参数，至少读取一个文件，第一个参数是程序名本身if len(os.Args) < 2 {// 如果参数数量小于2，打印到标准错误fmt.Fprintf(os.Stderr, "Usage: mrcoordinator inputfiles...\n")os.Exit(1)}// 创建协调者实例，除了程序名，剩下的参数作为输入文件传递m := mr.MakeCoordinator(os.Args[1:], 10) // 10 个工作者// 循环直到 MapReduce 作业完成for m.Done() == false { // m.Done() 检查 mr 作业是否完成// time.Sleep 暂停一秒time.Sleep(time.Second)}// 作业完成后，再等待一秒钟，可能是为了确保所有输出都已经写入time.Sleep(time.Second)
}

main/mrworker.go

从命令行参数中，获取插件文件，并将插件文件中的 Map 和 Reduce 函数，转化为具体函数类型（便于后续调用）

因为接口类型本身，不能直接被调用，需要转化为具体类型

package mainimport ("6.824/mr""plugin""os""fmt""log"
)// main 是程序的入口点，当程序启动时最先执行的函数
func main() {// 参数1：程序名  参数2：插件文件路径if len(os.Args) != 2 {// 写入标准错误流fmt.Fprintf(os.Stderr, "Usage: mrworker xxx.so\n")// 终止程序，并返回状态码 1 表示错误os.Exit(1)}// 调用 loadPlugin 函数加载 Map 和 Reduce 函数// Map, Reduce 函数，都来自于插件文件// mapf 是 Map 函数，reducef 是 Reduce 函数mapf, reducef := loadPlugin(os.Args[1])// 调用 mr.Worker 启动 MapReduce 工作者进程，传入加载的 Map 和 Reduce 函数mr.Worker(mapf, reducef)
}// loadPlugin 函数用于从插件文件中加载 Map 和 Reduce 函数
// filename 插件文件的路径
func loadPlugin(filename string) (func(string, string) []mr.KeyValue, func(string, []string) string) {// 使用 plugin.Open 函数打开插件文件，返回插件对象实例 p 和可能发生的错误 err// p 包含 Map 和 Reduce 函数p, err := plugin.Open(filename)if err != nil {log.Fatalf("cannot load plugin: %v", filename)}// p.Lookup 方法查找插件中名为 "Map" 的导出符号，其实就是 Map 函数// xmapf 是一个 plugin.Symbol 类型的变量，用于存储从插件中查找到的 Map 函数符号// plugin.Symbol 是一个接口类型，代表插件中的任意导出符号xmapf, err := p.Lookup("Map")if err != nil {log.Fatalf("cannot find Map function in plugin: %v", filename)}// 类型断言用于确定 xmapf 中存储的具体函数类型// 并将其从 plugin.Symbol 接口类型断言回其静态的函数类型// xmapf.() 就是类型断言, 将 plugin.Symbol 转化为具体函数类型mapf := xmapf.(func(string, string) []mr.KeyValue)// 同上，查找并断言 Reduce 函数xreducef, err := p.Lookup("Reduce")if err != nil {log.Fatalf("cannot find Reduce function in plugin: %v", filename)}reducef := xreducef.(func(string, []string) string)// 返回加载并断言成功的 Map 和 Reduce 函数return mapf, reducef
}

mr/coordinator.go

struct Coordinator：分配 MapReduce 任务到对应 worker

Example()：rpc 处理函数的例子

server()：启动 rpc 服务，监听来自 worker 的请求

Done()：检查 MapReduce 作业是否完成

MakeCoordinator()：创建并初始化 Coordinator 实例

package mrimport ("log""net""os""net/rpc""net/http"
)// Coordinator 负责管理和分配任务
type Coordinator struct {// Your definitions here.
}// RPC handlers for the worker to call.
// an example RPC handler.
// the RPC argument and reply types are defined in rpc.go
// rpc 调用的参数和返回值，在 rpc.go 中定义
func (c *Coordinator) Example(args *ExampleArgs, reply *ExampleReply) error {reply.Y = args.X + 1return nil // rpc 调用成功
}// start a thread that listens for RPCs from worker.go
func (c *Coordinator) server() {// 注册协调者实例，处理 RPC 调用rpc.Register(c)// 允许使用 HTTP 协议进行 RPC 通信rpc.HandleHTTP()// 协调者 socket 文件名sockname := coordinatorSock()// 监听前移除已存在的 socket 文件，避免监听失败os.Remove(sockname)// 监听 UNIX socket，准备接收来自 worker 的连接l, e := net.Listen("unix", sockname)if e != nil {log.Fatal("listen error:", e)}// 新的 goroutine 中启动 HTTP 服务，以处理 RPC 请求go http.Serve(l, nil)
}// main/mrcoordinator.go 会定期调用 Done() 函数来检查整个作业是否已完成。
func (c *Coordinator) Done() bool {ret := false// Your code here to implement the check for completion of all tasks// 在这里实现检查所有任务是否完成的逻辑，例如检查所有 Map 和 Reduce 任务的状态return ret // 作业是否完成
}// create a Coordinator
// main/mrcoordinator.go calls this function
// nReduce is the number of reduce tasks to use.
// The returned value is a pointer to the newly created Coordinator instance.
func MakeCoordinator(files []string, nReduce int) *Coordinator {c := Coordinator{}// Your code here to initialize the Coordinator, e.g., load input files, setup tasks, etc// 启动 RPC 服务器线程，以便监听和处理来自 worker 的 RPC 请求c.server()// 返回指向新创建的协调者实例的指针，这样调用者就可以通过这个指针来访问和操作协调者实例return &c
}

mr/worker.go

KeyValue 结构体

ihash()：返回 reduce 任务编号（用于发送 Map 输出的数据）

Worker()：调用插件中的 map() 和 reduce() 函数

CallExample()：展示 rpc 调用的完整流程，需要借助 call()

call()：建立 rpc 连接，再发送 rpc 请求

// package mr - 定义了MapReduce作业的工作者包，包含实现MapReduce算法所需的结构和函数// import语句 - 日志记录、rpc远程过程调用、哈希计算package mrimport ("fmt""log""net/rpc""hash/fnv"
)// 定义 MapReduce 中的键值对
type KeyValue struct {Key   stringValue string
}// 自定义的哈希函数，用于确定Map输出的键值对，应该发送到哪个Reduce任务
// Map阶段输出的键分配到不同的Reduce任务
func ihash(key string) int {h := fnv.New32a() // 创建FNV-1a哈希生成器// 字符串 key 转为 []byte 字节切片，因为 Wirte() 需要操作字节数据h.Write([]byte(key)) // 将键的字节序列写入哈希生成器// 使用按位与操作确保结果是一个非负整数，适合作为索引使用// 0x7fffffff 就是 0111 1111 ... 1111，符号位为正，其他不变return int(h.Sum32() & 0x7fffffff)
}// Worker 函数 - 是MapReduce工作者的主要工作函数
// 它调用用户提供的map和reduce函数
// main/mrworker.go calls this function.
// 传入的两个参数是 mapf() 和 reducef()
func Worker(mapf func(string, string) []KeyValue, reducef func(string, []string) string) {// 工作者实现细节将在这里编写，包括从协调者接收任务和发送结果
}// example function to show how to make an RPC call to the coordinator.
func CallExample() {// {X: 99} 结构体字面量, X 初始化为 99args := ExampleArgs{X: 99} // rpc通信中传递的参数reply := ExampleReply{}  // 用于存储响应的返回值// 发送RPC请求到协调者，等待回复// 服务名称.方法名称，rpc包会根据这个字符串，找到对应的服务和方法进行调用call("Coordinator.Example", &args, &reply)fmt.Printf("reply.Y %v\n", reply.Y) 
}// send an RPC request to the coordinator, wait for the response.
func call(rpcname string, args interface{}, reply interface{}) bool {sockname := coordinatorSock() // 获取协调者socket名称c, err := rpc.DialHTTP("unix", sockname) // 建立RPC连接if err != nil {log.Fatal("dialing:", err)}defer c.Close()// Call 方法是 net/rpc 包中的 *rpc.Client 类型的一个实例方法err = c.Call(rpcname, args, reply) // 发送RPC请求if err == nil {return true}fmt.Println(err)return false
}

mr/rpc.go

ExampleArgs 和 ExampleReply，表示 rpc 参数和 rpc 返回值两种类型

coordinatorSock()：为协调者生成 socket 文件名

package mr// RPC definitions
// remember to capitalize(大写) all namesimport "os"  // 操作系统功能，获取用户ID
import "strconv"  // 字符串转换// example to show how to declare the arguments(参数)
// and reply(返回值) for an RPCtype ExampleArgs struct {X int 
}type ExampleReply struct {Y int 
}// Add your RPC definitions here// Cook up a unique-ish UNIX-domain socket name
// in /var/tmp, for the coordinator// Can't use the current directory since
// Athena AFS doesn't support UNIX-domain sockets.// 这里指定的是一个UNIX域socket的文件路径前缀，它位于/var/tmp目录下
// 并且以"824-mr-"作为前缀，以确保socket文件名的唯一性
// 用于获取协调者的socket文件名，以便建立RPC连接
func coordinatorSock() string {// 定义UNIX域socket的基础路径，前缀为"/var/tmp/824-mr-"s := "/var/tmp/824-mr-"  // 将当前用户的UID转换为字符串并追加到基础路径之后，创建一个唯一的socket文件名s += strconv.Itoa(os.Getuid())  return s  // 协调者监听的socket文件的路径
}