挺早就听闻MIT6.824的大名，趁着这段时间比较闲，一边学习一边记录一下实验的完成过程。

MapReduce#

MapReduce

MapReduce是一个用于大规模数据集并行处理的分布式计算框架。

步骤：

从分布式文件系统中加载文件并进行分片 ( Split ) .
Master 节点通过心跳机制检测Worker节点状态，并分配任务给 Worker 节点。
Worker 节点执行 Map 任务，将原始分配文件转换为中间键值对。随后，这些键值对会经过分区并在 shuffle 阶段进行分组和排序。
Map 阶段结束后，Reduce Worker 接收分组后的键值对数据，执行 Reduce 操作并将最终结果文件输出到分布式文件系统中。

Lab1#

Lab1的要求是在提供的code框架基础上，实现worker节点和Coordinator ( Master ) 节点的基本功能。

总体难度不高，思路也比较清晰。如果之前没怎么接触过go建议先看一下课程的LEC 5了解golang的并发设计方式。

Worker#

对于 worker 节点，按照 hints 里的建议，首先修改Worker()，向Master节点发送RPC请求任务，得到任务后根据任务类型进行不同的操作。实验要求当没有任务时Worker应该退出，这里我直接采用了 reply 中没有任务时Worker 自行结束的方案。

1
// main/mrworker.go calls this function.
2
func Worker(mapf func(string, string) []KeyValue,
3
  reducef func(string, []string) string) {
4
  // Your worker implementation here.
5
  // uncomment to send the Example RPC to the coordinator.
6
  // CallExample()
7
  for {
8
    args := Args{}
9
    reply := Reply{}
10
    call("Coordinator.RPCMaster", &args, &reply)
11
    if reply.MapTask != nil {
12
      doMapTask(mapf, reply.MapTask, reply.NReduce)
13
    } else if reply.ReduceTask != nil {
14
      doReduceTask(reducef, reply.ReduceTask, reply.FilesLen)
15
    } else {
16
      return
17
    }
18
  }
19
}

Map阶段#

1
func doMapTask(mapf func(string, string) []KeyValue, task *MapTask, nReduce int) error {
2
  file, err := os.Open(task.Filename)
3
  if err != nil {
4
    log.Fatalf("cannot open %v", task.Filename)
5
  }
6
  content, err := ioutil.ReadAll(file)
7
  if err != nil {
8
    log.Fatalf("cannont read %v", task.Filename)
9
  }
10
  file.Close()
11

12
  kva := mapf(task.Filename, string(content))
13
  sort.Sort(ByKey(kva))

文件操作我们可以直接 copy mrsequential.go 中提供的部分。重点是分区操作，需要根据键的哈希值将键值对分配到对应的 Reduce 分区。

1
omap := make([][]KeyValue, nReduce)
2
  for _, kv := range kva {
3
    reduceNum := ihash(kv.Key) % nReduce
4
    omap[reduceNum] = append(omap[reduceNum], kv)
5
  }
6
  for i := 0; i < nReduce; i++ {
7
    intermediateFileName := fmt.Sprintf("map-%d-%d.json", task.Num, i)
8
    tempFile, err := ioutil.TempFile(".", "tmp-")
9
    tempFileName := tempFile.Name()
10
    if err != nil {
11
      return fmt.Errorf("failed to create temp file: %w", err)
12
    }
13
    enc := json.NewEncoder(tempFile)
14
    for _, kv := range omap[i] {
15
      err := enc.Encode(&kv)
16
      if err != nil {
17
        return fmt.Errorf("failed to create json file: %w", err)
18
      }
19
    }
20
    tempFile.Close()
21
    if err := os.Rename(tempFileName, intermediateFileName); err != nil {
22
      return fmt.Errorf("failed to rename temp file to target file: %w", err)
23
    }
24
  }

omap 数组存储本次 Map 结果不同键应处的桶，创建对应文件并写入。最后，我们需要让 Master 知道这个节点的任务已经完成，所以我们还需要定义另一个 RPC 函数。

1
    args := task
2
    var reply Reply
3
    call("Coordinator.MapTaskComplete", &args, &reply)
4

5
    return nil
6
  }

Reduce阶段#

Reduce 同理，其实实现 Worker 部分的注意点在实验 hints 的讲述都比较详细了，具体的实现也写在了 mrsequential.go 中。

1
func doReduceTask(reducef func(string, []string) string, task *ReduceTask, fileLens int) error {
2
  var kva []KeyValue
3
  for i := 0; i < fileLens; i++ {
4
    intermediateFileName := fmt.Sprintf("map-%d-%d.json", i, task.Num)
5
    file, err := os.Open(intermediateFileName)
6
    if err != nil && !os.IsNotExist(err) {
7
      return fmt.Errorf("Error opening file: %v\n", err)
8
    }
9
    dec := json.NewDecoder(file)
10
    for {
11
      var kv KeyValue
12
      if err := dec.Decode(&kv); err != nil {
13
        break
14
      }
15
      kva = append(kva, kv)
16
    }
17
  }
18
  sort.Sort(ByKey(kva))

收集任务 id 对应的 Map 任务的中间结果文件，然后 copy 一下 mrsequential.go 中具体的归约操作。

1
tempFile, err := ioutil.TempFile(".", "tmp-")
2
  tempFileName := tempFile.Name()
3
  if err != nil {
4
    return fmt.Errorf("failed to create temp file: %w", err)
5
  }
6
  oname := fmt.Sprintf("mr-out-%d", task.Num)
7
  i := 0
8
  for i < len(kva) {
9
    j := i + 1
10
    for j < len(kva) && kva[j].Key == kva[i].Key {
11
      j++
12
    }
13
    values := []string{}
14
    for k := i; k < j; k++ {
15
      values = append(values, kva[k].Value)
16
    }
17
    output := reducef(kva[i].Key, values)
18

19
    fmt.Fprintf(tempFile, "%v %v\n", kva[i].Key, output)
20
    i = j
21
  }
22
  tempFile.Close()
23
  if err := os.Rename(tempFileName, oname); err != nil {
24
    return fmt.Errorf("failed to rename temp file to target file: %w", err)
25
  }
26
  args := task
27
  var reply Reply
28
  call("Coordinator.ReduceTaskComplete", &args, &reply)
29
  return nil
30
}

最后，和 Map 阶段一致，我们也需要 RPC 通知 Master 节点任务完成。

Coordinator#

我使用的 Master 结构体：

1
type Coordinator struct {
2
  // Your definitions here.
3
  mapCompletedTaskNums    int
4
  reduceCompletedTaskNums int
5
  nReduce                 int
6
  mapTaskCompleted        bool
7
  reduceTaskCompleted     bool
8
  mapTasks                []*MapTask
9
  reduceTasks             []*ReduceTask
10
  mu                      sync.Mutex
11
  cond                    *sync.Cond
12
}

思路就是 Master 维护两个任务列表，在 Worker 请求任务时将任务分配过去并监测任务的状态。任务失败则改回任务状态等待下一个来请求任务的节点，任务成功则更改任务状态为成功并检查所有的阶段任务 ( Map/Reduce ) 是否全部完成。

（代码写得不怎么优雅，见谅）

RPC应答函数：

1
func (c *Coordinator) RPCMaster(args *Args, reply *Reply) error {
2
  c.mu.Lock()
3
  defer c.mu.Unlock()
4
  for { //map
5
    if c.mapTaskCompleted {
6
      break
7
    } else if task := c.fetchMapTask(); task != nil {
8
      reply.MapTask = task
9
      reply.NReduce = c.nReduce
10
      c.mapTaskStart(task)
11
      return nil
12
    } else {
13
      c.cond.Wait()
14
    }
15
  }
16
  for { //reduce
17
    if c.reduceTaskCompleted {
18
      break
19
    } else if task := c.fetchReduceTask(); task != nil {
20
      reply.ReduceTask = task
21
      reply.FilesLen = len(c.mapTasks)
22
      c.reduceTaskStart(task)
23
      return nil
24
    } else {
25
      c.cond.Wait()
26
    }
27
  }
28
  return nil
29
}

任务开始函数：

1
func (c *Coordinator) mapTaskStart(task *MapTask) {
2
  task.State = STARTED
3
  go func(task *MapTask) {
4
    timedue := time.After(10 * time.Second)
5
    <-timedue
6
    c.mu.Lock()
7
    defer c.mu.Unlock()
8
    if task.State != FINISHED {
9
      log.Printf("recover map task %d \n", task.Num)
10
      task.State = WAITTING
11
      c.cond.Broadcast()
12
    }
13
  }(task)
14
}

计时器检查任务状态，若失败则改回任务状态并通知RPC协程停止阻塞。

任务完成函数：

1
func (c *Coordinator) MapTaskComplete(task *MapTask, reply *Reply) error {
2
  c.mu.Lock()
3
  defer c.mu.Unlock()
4
  c.mapTasks[task.Num].State = FINISHED
5
  c.mapCompletedTaskNums++
6
  if c.mapCompletedTaskNums == len(c.mapTasks) {
7
    c.mapTaskCompleted = true
8
    c.cond.Broadcast()
9
  }
10
  return nil
11
}

在Worker完成任务后RPC调用此函数更改任务状态。