Map阶段#

func doMapTask(mapf func(string, string) []KeyValue, task *MapTask, nReduce int) error {
	file, err := os.Open(task.Filename)
	if err != nil {
		log.Fatalf("cannot open %v", task.Filename)
	}
	content, err := ioutil.ReadAll(file)
	if err != nil {
		log.Fatalf("cannont read %v", task.Filename)
	}
	file.Close()

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

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

omap := make([][]KeyValue, nReduce)
	for _, kv := range kva {
		reduceNum := ihash(kv.Key) % nReduce
		omap[reduceNum] = append(omap[reduceNum], kv)
	}
	for i := 0; i < nReduce; i++ {
		intermediateFileName := fmt.Sprintf("map-%d-%d.json", task.Num, i)
		tempFile, err := ioutil.TempFile(".", "tmp-")
		tempFileName := tempFile.Name()
		if err != nil {
			return fmt.Errorf("failed to create temp file: %w", err)
		}
		enc := json.NewEncoder(tempFile)
		for _, kv := range omap[i] {
			err := enc.Encode(&kv)
			if err != nil {
				return fmt.Errorf("failed to create json file: %w", err)
			}
		}
		tempFile.Close()
		if err := os.Rename(tempFileName, intermediateFileName); err != nil {
			return fmt.Errorf("failed to rename temp file to target file: %w", err)
		}
	}

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

		args := task
		var reply Reply
		call("Coordinator.MapTaskComplete", &args, &reply)
	
		return nil
	}

Reduce阶段#

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

func doReduceTask(reducef func(string, []string) string, task *ReduceTask, fileLens int) error {
	var kva []KeyValue
	for i := 0; i < fileLens; i++ {
		intermediateFileName := fmt.Sprintf("map-%d-%d.json", i, task.Num)
		file, err := os.Open(intermediateFileName)
		if err != nil && !os.IsNotExist(err) {
			return fmt.Errorf("Error opening file: %v\n", err)
		}
		dec := json.NewDecoder(file)
		for {
			var kv KeyValue
			if err := dec.Decode(&kv); err != nil {
				break
			}
			kva = append(kva, kv)
		}
	}
	sort.Sort(ByKey(kva))

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

tempFile, err := ioutil.TempFile(".", "tmp-")
	tempFileName := tempFile.Name()
	if err != nil {
		return fmt.Errorf("failed to create temp file: %w", err)
	}
	oname := fmt.Sprintf("mr-out-%d", task.Num)
	i := 0
	for i < len(kva) {
		j := i + 1
		for j < len(kva) && kva[j].Key == kva[i].Key {
			j++
		}
		values := []string{}
		for k := i; k < j; k++ {
			values = append(values, kva[k].Value)
		}
		output := reducef(kva[i].Key, values)

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

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