MIT6.824 mapReduce lab1 reduce过程实现
程序员文章站
2022-07-12 17:02:36
...
worker
1. worker像master请求reduce任务,master分配reduce任务给worker
2. 经过map过后的文件格式如下:mr-X-Y, where X is the Map task number, and Y is the reduce task number.
执行reduce过程时,需要对X遍历,然后对指定的Y reduce
3. reduce算法过程如下,使用一个map数据结构 interMaps := map[string][]string{}, 实现算法如下:
// 收集所有的key string和value []string在map中,然后传递给worker的reducef中
interMaps := map[string][]string{}
interFileNames := []string{}
for i := 0; i < nMapTask; i++ {
interFileName := fmt.Sprintf("mr-%d-%d", i, reduceTaskID)
interFileNames = append(interFileNames, interFileName) //留着之后删除中间结果用
// 打开文件
file, err := os.Open(interFileName)
if err != nil {
log.Fatalf("error: Can't open interFile: %v", interFileName)
}
// decoder读取信息放入interMaps中
decoder := json.NewDecoder(file)
for {
var kv KeyValue // 解码后的结果存在kv中
if err := decoder.Decode(&kv); err != nil {
if err == io.EOF {
break
} else {
log.Fatal(err)
}
}
// 如果Key还没有被统计过
if _, ok := interMaps[kv.Key]; !ok {
interMaps[kv.Key] = []string{kv.Value}
} else {
// 被统计过了
interMaps[kv.Key] = append(interMaps[kv.Key], kv.Value)
}
}
}
// 遍历interMaps使用reducef函数(参数是string,[]string)得到最终的结果,然后写到临时文件中
reduceResult := []string{}
tmpFileName := make([]string, 1)
finalFileName := make([]string, 1)
for k, v := range interMaps {
reduceResult = append(reduceResult, fmt.Sprintf("%v %v\n", k, w.reducef(k, v)))
}
然后就是一些写文件的过程。
上一篇: ucore lab1
下一篇: 微服务整合Seata分布式事务处理