6.824-Lab1: MapReduce

vickeex · · 923 次点击 · · 开始浏览    
这是一个创建于 的文章,其中的信息可能已经有所发展或是发生改变。

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(): 阻塞,直至所有任务完成

有疑问加站长微信联系(非本文作者)

本文来自:简书

感谢作者:vickeex

查看原文:6.824-Lab1: MapReduce

入群交流(和以上内容无关):加入Go大咖交流群,或添加微信:liuxiaoyan-s 备注:入群;或加QQ群:692541889

923 次点击  
加入收藏 微博
暂无回复
添加一条新回复 (您需要 登录 后才能回复 没有账号 ?)
  • 请尽量让自己的回复能够对别人有帮助
  • 支持 Markdown 格式, **粗体**、~~删除线~~、`单行代码`
  • 支持 @ 本站用户;支持表情(输入 : 提示),见 Emoji cheat sheet
  • 图片支持拖拽、截图粘贴等方式上传