欢迎您访问程序员文章站本站旨在为大家提供分享程序员计算机编程知识!
您现在的位置是: 首页

6.824 Lab 1: MapReduce

程序员文章站 2022-07-12 17:41:13
...

计划过很多次,终于开始了6.824的征程;

希望一鼓作气!

一: MapReduce逻辑

6.824 Lab 1:MapReduce 代码逻辑图

二: 实验任务

  1. 完成用户端map()+reduce()函数。调用MapReduce接口,用于单词统计,倒排索引生成;
  2. 完成MapReduce端任务处理函数doMap()+doReduce(),一个Map/Reduce的任务具体的执行流程;
  3. 完成MapReduce端调度函数schedule(),master如何调度worker进行分布式作业;
  4. 完成MapReduce端fault tolerance,处理worker failures;

三: 实验问题

Part1:Map/Reduce input and output
json编解码需要一致
//merge阶段,读取每个reduce任务输出的文件,一个个解析;
dec := json.NewDecoder(file)
for {
    var kv KeyValue
    err = dec.Decode(&kv)
    if err != nil {
    break
    }
    kvs[kv.Key] = kv.Value
}
//原始版本:reduce阶段,写入文件,将切片整体写入,导致merge无法解析,最终check不正确;
var values []KeyValue   
enc := json.NewEncoder(file)
err:=enc.Encode(&values)

//修改版本:reduce阶段,写入文件,一个个写入;
enc := json.NewEncoder(file)
for key,values := range keys {
    enc.Encode(KeyValue{key, reduceF(key,values)})
}
Part3: Distributing MapReduce tasks
//workers资源池
workers := make(chan string, ntasks)
go func() {
  for workerAdress := range registerChan {
    workers <- workerAdress
  }
  //原有错误代码
  // workerAdress := <-registerChan
  // workers <- workerAdress
}()
    
//错误原因:map/reduce阶段只接受一个worker;导致另一个worker工作任务为0;
//当map阶段和reduce阶段接受的worker不同时,偶尔返回正确答案,但也是错误的,因为每个阶段仍然是只有一个worker在处理;
注意并发参数
func main() {
    var wg sync.WaitGroup
    wg.Add(10)
    for i := 0; i < 10; i++ {
        go func() {
            fmt.Println(i)
            wg.Done()
        }
    }
    wg.Wait()
}
10
6
6
10
10
10
10
10
10
4
Part4: Handling worker failures
(1)
func work(workers chan string, doTaskArgs DoTaskArgs) {
    defer wg.Done()
    Here:
    workerAdress := <-workers //获取工作线程
    if call(workerAdress, "Worker.DoTask", doTaskArgs, nil) {
        workers <- workerAdress //工作正常完成
    }else{
        workers <- workerAdress //工作非正常完成
        goto Here
    }
}

(2)
func work(workers chan string, doTaskArgs DoTaskArgs) {
    defer wg.Done()
    Here:
    workerAdress := <-workers //获取工作线程
    if call(workerAdress, "Worker.DoTask", doTaskArgs, nil) {//工作正常完成
        workers <- workerAdress 
    }else{//工作非正常完成
        // workers <- workerAdress 
        goto Here
    }
}

崩溃的worker不需要再放入workers资源池中,worker的nRPC为0时,会断开链接;
for {
        wk.Lock()
        if wk.nRPC == 0 {
            wk.Unlock()
            break
        }
        wk.Unlock()
        conn, err := wk.l.Accept()
        if err == nil {
            wk.Lock()
            wk.nRPC--
            wk.Unlock()
            go rpcs.ServeConn(conn)
        } else {
            break
        }
    }
wk.l.Close()
debug("RunWorker %s exit\n", me)
Part5: Inverted index generation
题目的文件输出是按照文件名排序,并没有按照文件的次数排序;
func reduceF(key string, values []string) string {
    // TODO: you should complete this to do the inverted index challenge
    //每篇文章的数目
    documentNumMap := make(map[string]int)
    for _, document := range values {
        documentNumMap[document]++
    }
    // //相同次数的文章集合
    // numDocumentMap := make(map[int][]string)
    // for document, number := range documentNumMap {
    //  numDocumentMap[number] = append(numDocumentMap[number], document)
    // }
    // //根据key排序
    // documentNums := make([]int, 0)
    // for number, _ := range numDocumentMap {
    //  documentNums = append(documentNums, number)
    // }
    // sort.Ints(documentNums)

    // //输出
    // documents:=make([]string,0)
    // for index := 0; index < len(documentNums); index++ {
    //  documentNum := documentNums[index]
    //  documents = append(documents,numDocumentMap[documentNum]...)
    // }
    // return strconv.Itoa(len(documentNumMap)) + " " + strings.Join(documents, ",")

    documents:=make([]string,0)
    for document,_:=range documentNumMap{
        documents = append(documents,document)
    }
    sort.Strings(documents)
    return strconv.Itoa(len(documentNumMap)) + " " + strings.Join(documents, ",")   
}
Test

实验每个part单独做测试时均正常运行,但是在最后整体测试时,part3、part4偶尔会出现报错:

error: listen unix /var/tmp/824-502/mr47307-worker13: socket: too many open files in system

//1:读取每个临时文件中的数据并整合到一起
keyValues := make([]KeyValue, 0)
for index := 0; index < nMap; index++ {
  fileName := reduceName(jobName, index, reduceTaskNumber)
  file, errFile := os.Open(fileName)
  if errFile != nil {
    log.Fatal(errFile)
  }
  defer file.Close()
  var tempKeyValues []KeyValue //一个文件中的数据,临时,否则会被覆盖
  errDecode := json.NewDecoder(file).Decode(&tempKeyValues)
  keyValues = append(keyValues, tempKeyValues...)
  if errDecode != nil {
    log.Fatal(errDecode)
  }
}
    
    
keyValues := make([]KeyValue, 0)
for index := 0; index < nMap; index++ {
  fileName := reduceName(jobName, index, reduceTaskNumber)
  file, err := os.Open(fileName)
  if err != nil {
    log.Fatal("doReduce: ",err)
  }
  dec:=json.NewDecoder(file)
  for{
    var kv KeyValue
    err=dec.Decode(&kv)
    if err!=nil {
      break;
    }
    keyValues = append(keyValues,kv)
  }
  file.Close()
}

defer 导致函数结束时才释放文件,nMap个文件全部open状态,这可能是报错的原因; 

修改完毕,全部测试pass;

Lab1 2017-09-10

四: 总结

  1. MapReduce
    • 概念理解
    • 实现理解
  2. 熟悉go语言
    • 实验需求-》学习动力
    • 调试BUG
    • 学习提供代码

[2017.9 梦工厂]