【MIT-6.824】Lab 1: MapReduce

Lab 1链接：https://pdos.csail.mit.edu/6.824/labs/lab-1.html

Part I: Map/Reduce input and output

 Part I需要补充两个关键功能：为map函数分解输出的功能和为reduce函数收集输入的功能，这两个功能对应的函数分别在common_map.go的doMap()函数和common_reduce.go的doRedce()函数。

本人首先梳理程序运行流程，其次补充代码，最后测试结果。

程序运行流程简述如下：

1. Sequential首先获取Master对象的指针，然后利用函数闭包运行Master.run()。
2. Master.run()会依次运行mapPhase和reducePhase。
3. 在mapPhase中，doMap会依次处理每一个输入文件；在reducePhase中，doReduce会依次处理nReduce（论文中为R）个区域。

为实现doMap函数，需要实现以下功能：

1. 读取inFile。
2. 通过mapF函数，将inFile转换成key/value的切片形式。
3. 将上一步得到的结果切割为nReduce个切片，并使用hash函数将结果分配到对应的切片中。
4. 将上一步得到的结果转换为Json格式，并存储于文件中。

func doMap(
    jobName string, // the name of the MapReduce job
    mapTask int, // which map task this is
    inFile string,
    nReduce int, // the number of reduce task that will be run ("R" in the paper)
    mapF func(filename string, contents string) []KeyValue,
) {    
    // Your code here (Part I).
    // read file
    data, err := ioutil.ReadFile(inFile)
    if err != nil{
        log.Fatal("common_map.doMap: fail to read the file. The error is ", err)
    }

    // transfer file
    slice := mapF(inFile, string(data))

    // initialize reduceKv
    var reduceKv [][]KeyValue
    for i := 0; i < nReduce; i++{
        temp := make([]KeyValue, 0)
        reduceKv = append(reduceKv, temp)
    }

    // get reduceKv
    for _, s := range slice{
        index := ihash(s.Key) % nReduce
        reduceKv[index] = append(reduceKv[index], s)
    }

    // get intermediate files
    for i:= 0; i < nReduce; i++{
        file, err := os.Create(reduceName(jobName, mapTask, i))
        if err != nil{
            log.Fatal("common_map.doMap: fail to create the file. The error is ", err)
        }
        enc := json.NewEncoder(file)
        for _, kv := range(reduceKv[i]){
            err := enc.Encode(&kv)
            if err != nil{
                log.Fatal("common_map.doMap: fail to encode. The error is ", err)
            }
        }
        file.Close()
    }
}

为实现doReduce函数，需要实现如下功能：

1. 读取文件中存储的key/value对，并对其进行排序。
2. 将key值相同的value发送至用户定义的reduceF()中，reduceF()会返回一个新的value值。
3. 将新的key/value对写入文件。

func doReduce(
    jobName string, // the name of the whole MapReduce job
    reduceTask int, // which reduce task this is
    outFile string, // write the output here
    nMap int, // the number of map tasks that were run ("M" in the paper)
    reduceF func(key string, values []string) string,
) {    
    // Your code here (Part I).

    // get and decode file
    var slices []KeyValue
    for i := 0; i < nMap; i++{
        fileName := reduceName(jobName, i, reduceTask)
        file, err := os.Open(fileName)
        if err != nil{
            log.Fatal("common_reduce.doReduce: fail to open the file. The error is ", err)
        }
        dec := json.NewDecoder(file)
        var kv KeyValue
        for{
            err := dec.Decode(&kv)
            if err != nil{
                break
            }
            slices = append(slices, kv)
        }
        file.Close()
    }

    sort.Sort(ByKey(slices))
    
    //return the reduced value for the key
    var reducedValue []string
    var outputValue []KeyValue
    preKey := slices[0].Key
    for i, kv := range slices{
        if kv.Key != preKey{
            outputValue = append(outputValue, KeyValue{preKey, reduceF(preKey, reducedValue)})
            reducedValue = make([]string,0)
        }
        reducedValue = append(reducedValue, kv.Value)
        preKey = kv.Key

        if i == (len(slices) - 1){
            outputValue = append(outputValue, KeyValue{preKey, reduceF(preKey, reducedValue)})
        }
    }

    //write and encode file
    file, err := os.Create(outFile)
    if err != nil{
        log.Fatal("common_reduce.doReduce: fail to create the file. The error is ", err)
    }
    defer file.Close()

    enc := json.NewEncoder(file)
    for _, kv := range outputValue{
        err := enc.Encode(&kv)
        if err != nil{
            log.Fatal("common_reduce.doReduce: fail to encode. The error is ", err)
        }
    }
}

 实验结果如下图所示：

Part II: Single-worker word count

Part II需要统计文档中每个单词出现的数目，需要实现的函数为wc.go中的mapF()和reduceF()函数。

mapF()函数需要将文件拆分为单词，并返回mapreduce.KeyValue的形式。reduceF()函数需要统计每一个Key对应的Value出现的数目，并以string的形式返回。

func mapF(filename string, contents string) []mapreduce.KeyValue {
    // Your code here (Part II).
    f := func(c rune) bool{
        return !unicode.IsLetter(c)
    }

    words := strings.FieldsFunc(contents, f)
    var result []mapreduce.KeyValue
    for _, word := range words{
        result = append(result, mapreduce.KeyValue{word,"1"})
    }
    return result
}

func reduceF(key string, values []string) string {
    // Your code here (Part II).
    sum := 0
    for _, value := range values{
        i, err := strconv.Atoi(value)
        if err != nil{
            log.Fatal("wc.reduceF: fail to convert. The error is ", err)
        }
        sum += i
    }
    return strconv.Itoa(sum)
}

实验结果如下图所示：

Part III: Distributing MapReduce tasks&&Part IV: Handling worker failures

Part III和Part IV需要将顺序执行的MapReduce框架并行化并处理worker异常。

本人分别介绍worker和master的执行流程。

worker：RunWorker()首先被调用，该函数创建新Worker并通过call()函数向Master.Register()发送RPC。

master：

1. 在master.go的Distributed()函数中，master通过startRPCServer()启动RPC服务器，然后利用函数闭包运行run()函数。
2. 在run()函数中，master会依次运行schedule(mapPhase)和schedule(reducePhase)。
3. 在schedule(phase)函数中，master会开启新协程运行forwardRegistrations()函数，然后运行Part III和Part IV需要实现的schedule.go中的schedule()函数。
4. 在介绍worker的执行流程时，本人提到worker会向Master.Register()发送RPC。在Register()函数中，master会将新的worker添加至mr.workers中并告知forwardRegistrations()出现了新的worker。
5. 在forwardRegistrations()函数中，master通过mr.workers的数目判断是否有新的worker。若有新的worker，master通过channel通知schedule.go的schedule()函数。
6. 在schedule()函数中，master负责为worker分配task。

为实现master对worker的调度，需要在schedule()函数中实现如下功能。

1. 通过sync.WaitGroup判断全部任务是否完成。
2. 通过registerChan判断是否有新的worker。若有，开启新协程为此worker分配任务。
3. 通过带有缓冲的channel输入任务序号，从channel中取出任务序号并分配给worker。若worker异常，则重新输入任务序号。
4. 通过call()函数向worker发送RPC。

func schedule(jobName string, mapFiles []string, nReduce int, phase jobPhase, registerChan chan string) {
    var ntasks int
    var n_other int // number of inputs (for reduce) or outputs (for map)
    switch phase {
    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)
", ntasks, phase, n_other)

    // All ntasks tasks have to be scheduled on workers. Once all tasks
    // have completed successfully, schedule() should return.
    //
    // Your code here (Part III, Part IV).
    //
    var wg sync.WaitGroup
    wg.Add(ntasks)

    taskChan := make(chan int, ntasks)
    for i := 0; i < ntasks; i++{
        taskChan <- i
    }

    go func(){
        for{
            ch := <- registerChan
            go func(address string){
                for{
                    index := <- taskChan
                    result := call(address, "Worker.DoTask", &DoTaskArgs{jobName, mapFiles[index], phase, index, n_other},new(struct{}))
                    if result{
                        wg.Done()
                        fmt.Printf("Task %v has done
", index)
                    }else{
                        taskChan <- index
                    }
                }
            }(ch)
        }
    }()
    wg.Wait()
    fmt.Printf("Schedule: %v done
", phase)
}

Part V: Inverted index generation (optional, does not count in grade)

Part V需要实现倒排索引，需要补充的函数为ii.go中的mapF()和reduceF()函数。

mapF()函数需要对输入文件中的单词进行分割，返回以单词为Key，以文件题目为Value的切片。

reduceF()函数需要对相同Key对应的全部Value去重并排序，统计Value的个数。

func mapF(document string, value string) (res []mapreduce.KeyValue) {
    // Your code here (Part V).
    f := func(c rune) bool{
        return !unicode.IsLetter(c)
    }
    words := strings.FieldsFunc(value, f)
    var result []mapreduce.KeyValue
    for _, word := range words{
        result = append(result, mapreduce.KeyValue{word, document})
    }
    return result
}

func reduceF(key string, values []string) string {
    // Your code here (Part V).
    fileName := make(map[string]bool)

    for _, value := range values{
        fileName[value] = true
    }

    num := 0
    var documents []string
    for key := range fileName{
        num += 1
        documents = append(documents, key)
    }
    sort.Strings(documents)
    
    var result string
    for i, file := range documents{
        if i >= 1{
            result += ","
        }
        result += file
    }
    return strconv.Itoa(num) + " " + result
}

实验结果如下图所示：

Running all tests