6.824-Lab1: MapReduce

MapReduce, 批处理的典型之一。主要思想即“分而治之”，将一大批数据（一个大任务）分成多个子任务，分别进行运算（同时）（map），再将运算结果合起来（reduce）。

master: 负责任务调度

mapper: 执行各个子任务，map运算

reducer: 执行结果汇总，reduce运算

例：在K/V的wordCount中，源数据为一个大文件，每个mapper负责一部分文件的count，mapper的运算结果(intermediate K/Vs)再交由reducer进行汇总(merge)。

本次实验中对MapReduce进行了基本的使用和简单的任务分发实现。

Part 1: Map/Reduce input and output

task: 完成Map/Reduce的工作流程doMap()和doReduce部分

• doMap()：读取输入的文件，创建nReduce个中间结果文件，调用mapF函数并将其计算的K/V结果通过hash散列映射到各中间结果文件。
• doReduce()：读取中间结果文件，将各K/V按K值存入一个字典变量kvMap，调用reduceF函数以对kvMap进行合并(wordCount无需额外操作)，并将reduceF的结果写入最终结果文件。
  重点 ：清楚操作流程，Go的文件操作

Part 2: 实现单节点的wordCount

task: 即实现task1中调用的mapF()和reduceF函数。

• mapF(): 将输入流(string)分割成单词并存入字典sliceKV{word: count}
  func mapF(filename string, contents string) []mapreduce.KeyValue {... ...}
• reduceF(): wordCount中的此函数无需额外操作。在doReduce()中将中间结果存入kvMap时即完成了合并。

Part3: distributing MapReduce tasks

task: 实现schedule.go文件，完成对多个worker进行map、reduce 任务的调度。

• registerChan: workers(RPC address)的channel，存储当前空闲可用的worker（registered worker）
• taskChan: 记录未完成或失败的任务，和sync.WaitGroup共同使用，以便重新执行失败任务。
  具体实现如下：

var ntasks int
    var n_other int // number of inputs (for reduce) or outputs (for map)
    switch phase {  // 确定当前工作阶段，执行map还是reduce调度
    case mapPhase:
        ntasks = len(mapFiles)
        n_other = nReduce
    case reducePhase:
        ntasks = nReduce
        n_other = len(mapFiles)
    }

    fmt.Printf("Schedule: %v %v tasks (%d I/Os)\n", ntasks, phase, n_other)

    var wg sync.WaitGroup
    taskChan := make(chan int)
    go func() {
        for t := 0; t < ntasks; t++ {
            wg.Add(1)
            taskChan <- t
        }
        wg.Wait()  // 直至ntasks个任务全完成
        close(taskChan)
    }()

    for task := range taskChan {
        addr := <-registerChan  // 取一空闲worker
        go func(j int, addr string) {  // 通过RPC调用执行任务
            result := call(addr, "Worker.DoTask", DoTaskArgs{jobName, mapFiles[j], phase, j, n_other}, nil)
            fmt.Println("46 current: ", j)
            if result {
                wg.Done()
            } else {
                taskChan <- j  // 该任务失败，需再次放入taskChan以便重新执行，直至成功
            }
            registerChan <- addr  // 执行完任务的worker放回registerChan，以继续执行其他任务 
        }(task, addr)
    }
    wg.Wait()

    fmt.Printf("Schedule: %v done\n", phase)
}

重点：使用goroutine+channel+waitgroup进行协程同步和通信

• goroutine: golang的性能保障，用户态的调度粒度，每个goroutine在用户内存中有自己的栈空间。
• channel: 传递数据的管道，常用于存储各种中间结果，在复杂的逻辑处理中实用得不得了。
• sync.Waitgroup: 等待一组协程的完成。类似任务队列的数据结构，任务完成会被移出，否则会阻塞当前执行线程。提供以下三个操作:
  • Add(): 等待的goroutine加一
  • Done(): 有一个任务完成，等待的goroutine减一
  • Wait(): 阻塞，直至所有任务完成