Part I: Map/Reduce input and output

添加代码:

  • common_map.go: doMap()
  • common_reduce.go: doReduce()

完成后测试:

cd "<project_position>/src/mapreduce"go test -run Sequential

我们需要事先掌握如下:

  • 保存读取json文件
  • 文件输入输出.
  • defer语句, 用于释放资源.

common_map.go: doMap()

要干什么?

  1. doMap()管理一个Map Task.
  2. 调用mapF()将输入文件的内容变成一个[]KeyValue.
  3. 用划分函数将mapF()的输出划分到nReduce个中间文件.
  4. 每个中间文件命由<JobName>-<mapTask>-<reduceTask>三部分组成.

代码构思:

  1. 创建nReduce个json文件, 将其句柄放入jsonArr这个数组中, 其类型为[][]KeyValue.
  2. 读取文件将其内容转为String, 输入mapF()得到输出kvList.
  3. 对于kvList中每个kv, 用index = ihash(kv.Key) % nReduce来将其加入数组jsonArr[index]中.
  4. 把jsonArr每一个数组jsonArr[i]保存到文件jobName-mapTask-i中(调用reduceName(..)).
  5. 中间文件的格式由自己设计, 只要doReduce()能够正常decode就好.

实现:

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,) {    contentsByte, err := ioutil.ReadFile(inFile)    if err != nil {        log.Fatal(err)    }    contents := string(contentsByte)    kvList := mapF(inFile, contents)    jsonArr := make([][]KeyValue, nReduce)    for _, kv := range kvList {        index := ihash(kv.Key) % nReduce // Key or Value ?        jsonArr[index] = append(jsonArr[index], kv)    }    // Save as json file.    for i := 0; i < nReduce; i++ {        filename := reduceName(jobName, mapTask, i)        file, err := os.Create(filename)        if err != nil {            log.Fatal(err)        }        enc := json.NewEncoder(file)        ok := enc.Encode(&jsonArr[i])         if ok != nil {            log.Fatal(ok)        }        file.Close()    }}

common_reduce.go: doReduce()

要干什么?

  1. doReduce()管理一个reduce任务r.
  2. 从每一个MapTask的ReduceTask编号为r的中间文件中(共nMap个), 读取kv对, 转为String.
  3. 将kv对按照key排序.
  4. 对每个不同的key调用reduceF(), 将输出结果写入磁盘

调用sort包来排序, 教程.

实现

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,) {    table := make(map[string][]string)    // Read nMap immediate files.    for i := 0; i < nMap; i++ {        file, err := os.Open(reduceName(jobName, i, reduceTask))        if err != nil {            log.Fatal(err)        }        enc := json.NewDecoder(file)        var tmp []KeyValue        decErr := enc.Decode(&tmp)        if decErr != nil {            log.Fatal(decErr)        }        for _, kv := range tmp {            table[kv.Key] = append(table[kv.Key], kv.Value)        }        file.Close()    }    // Sort them by Key    sortedKeyArr := make([]string, 0)    for k := range table {        sortedKeyArr = append(sortedKeyArr, k)    }    sort.Strings(sortedKeyArr)    outputFile, err := os.Create(outFile)    if err != nil {        log.Fatal(err)    }    defer outputFile.Close()    outputJSON := json.NewEncoder(outputFile)    for i := 0; i < len(sortedKeyArr); i++ {        key := sortedKeyArr[i]        val := table[key]        kv := KeyValue{key, reduceF(key, val)}        encErr := outputJSON.Encode(kv)        if encErr != nil {            log.Fatal(encErr)        }    }}

可通过测试.