实现一个MapReduce

开始学习分布式。

在这个练习中，我们将会使用 Go 实现一个叫作MapReduce的编程模型。

前言

在开始阅读本文之前，笔者希望你已经看过Google的论文MapReduce，本文只是对其的一个简单实现。当然，如果你的英文不是很好，也可以参照国内的一些翻译，感谢他们作出的贡献！

此外，mit的6.824课程也借用了这个模型作为lab1，一些其他的资源（代码和测试命令）在mit6.824上。

本文只是对整体的设计和思考做了些简述，代码由于课程的缘故不方便公开在GitHub上，以关键代码的形式放在文章中。

简述

MapReduce最早是由Google公司研究提出的一种面向大规模数据处理的并行计算模型和方法。其设计初衷主要是为了解决其搜索引擎中大规模网页数据的并行化处理，类似计算机中常见的分治思想。

首先假设有大量数据，我们希望统计数据当中每个单词出现的次数，结果保存在一个文件中。一个常见的思路是：首先将所有数据中的单词分割——即Map操作，然后排序并统计出现次数——即Reduce操作。但显然的问题是：

1.处理效率低下，一台计算机要进行大量数据的运算。

2.内存不足，需要读入大量数据分割，常见的排序难以满足要求。

因此，MapReduce应运而生。在这个模型下，有许多worker执行Map或Reduce操作，有一个Master负责调配整个系统的工作。具体流程如下图所示：

pic1

通过将Map调用的输入数据自动分割为M个数据片段的集合，Map调用被分布到多台机器上执行（其中每个集合中的数据以key-value形式保存，在这个案例下value通常为1，表示单词出现一次）。输入的数据片段能够在不同的机器上并行处理。使用分区函数将Map调用产生的中间key值分成R个不同分区（例如，hash(key) mod R），Reduce调用也被分布到多台机器上执行。分区数量（R）和分区函数由用户来指定。最终生成结果文件。

1
2
map     (k1,v1)       ->list(k2,v2) 
reduce  (k2,list(v2)) ->list(v2)

论文的具体内容在此不作赘述，接下来详述代码实现。

代码实现

Map和Reduce的具体操作代码在lab1中已经给出，我们只需要完成mr文件夹下的master.go、worker.go和rpc.go代码。

一.master.go

1.需要的变量类型

首先需要对整个系统的流程分阶段，其中Wait指Map任务都已下发但未全部提交成功。

1
const MapPhase, ReducePhase, Wait, Finish = 0, 1, 2, 3

其次是会用到的结构体，采用Master作为整个系统的调配对象，MapTask表示每个Map任务，ReduceTask表示每个Reduce任务。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
type Master struct {
	Phase int
	MappersChan chan MapTask
	ReducersChan chan ReduceTask

    NMapper int
    NReducer int

    FinishedMapper int
    FinishedReducer int

    MapOutPaths [][]string
    ReduceOutPaths []string

    mutex sync.Mutex

}

type MapTask struct {
	Infile string
	No int
}

type ReduceTask struct {
	Infiles []string
	No int
}

我们采用通道的数据结构（队列也可以）来传递下发任务，同时对系统进行一定的加锁处理。

2.关键的方法

主要为发送任务和报告完成的方法，均由worker主动调用。

发送任务
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
func (m *Master) SendTask(args *RPCArgs, reply *RPCReply) error {

	//mutex the phase
	m.mutex.Lock()
	reply.Phase = m.Phase
	m.mutex.Unlock()
	
	switch reply.Phase {
	case MapPhase:
		reply.MapTask = <- m.MappersChan
		reply.NReducer = m.NReducer
		//avoid crash
		go m.MapTimeout(reply.MapTask)
	case ReducePhase:
		reply.ReduceTask = <- m.ReducersChan
		go m.ReduceTimeout(reply.ReduceTask)
	}
	
	return nil

}

func (m *Master) MapTimeout(mTask MapTask) {
	time.Sleep(time.Second * 10)

	if len(m.MapOutPaths[mTask.No - 1]) == 0 {
		fmt.Printf("Mapper No.%v crash!\n", mTask.No)
		m.Phase = MapPhase
		m.MappersChan <- mTask
	}

}

func (m *Master) ReduceTimeout(rTask ReduceTask) {
	time.Sleep(time.Second * 10)

	if len(m.ReduceOutPaths[rTask.No - 1]) == 0 {
		fmt.Printf("Reducer No.%v crash!\n", rTask.No)
		m.ReducersChan <- rTask
	}

}

由于在lab1的第四个crash测试中，每个worker可能会出现异常退出或超时处理（事实上现实情况下分布式也很容易出现这种分区错误）。为保证这种分区容错性，我们对每个申请任务的worker都开启一个协程用以监视该任务是否完成（等待十秒），如果十秒过后还显示该任务未完成，master会将对应的任务重新加入到任务通道中。

报告完成
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
func (m *Master) ReportTask(args *RPCArgs, reply *RPCReply) error {
	m.mutex.Lock()
	defer m.mutex.Unlock()
	switch args.Phase {
	case MapPhase:
		//to avoid repeating report

		if len(m.MapOutPaths[args.No - 1]) != 0 {
			return nil
		}
	
		//add single OutPaths -> MapOutPaths
		for i := range args.OutPaths {
			m.MapOutPaths[args.No - 1] = append(m.MapOutPaths[args.No - 1], args.OutPaths[i])
		}
		m.FinishedMapper++
		//MappersChan is empty
		if len(m.MappersChan) == 0 {
			m.Phase = Wait
		}
	
		//MapPhase over
		if m.NMapper == m.FinishedMapper {
			m.shuffle()
			m.Phase = ReducePhase
		}
	
	case ReducePhase:
	
		if len(m.ReduceOutPaths[args.No - 1]) != 0 {
			return nil
		}
	
		m.ReduceOutPaths[args.No - 1] = args.OutPath
		m.FinishedReducer++
		//ReducePhase over
		if m.NReducer == m.FinishedReducer {
			m.Phase = Finish
			fmt.Println("master finished.")
		}
	}
	
	return nil

}

//MapPhase -> ReducePhase
func (m *Master) shuffle() {
	for i := 0; i < m.NReducer; i++ {
		task := ReduceTask {
			No: i + 1,
		}
		for j := 0; j < m.NMapper; j++ {
			task.Infiles = append(task.Infiles, m.MapOutPaths[j][i])
		}
		m.ReducersChan <- task
	}
}

这里有两个值得注意的地方：

①为保障上文所述的分区容错性，我们采取了相应的措施。但随之带来一个问题：假设上一个worker并没有崩溃，只是速度过慢呢？当超时的任务被重新加入通道，而上一次的任务结果发回master。最终就会有两份相同的任务结果传回master。笔者采取的应对措施是只保留最先到达的。

②在lab1的第三个reduce parallelism测试中，需要保证各worker对于reduce的处理是并行的。但情况并非如此：因为我们采用的是通道。

一个比较清晰的例子是：假设我们现在有很多台worker，有一台因老化而运行速度过慢（事实上只要速度不完全一致就会出现这种情况）。很容易出现这样的场景：很多worker提交了MapTask，直到剩余一个任务未完成。此时MapTask通道为空，所有进入的worker都会阻塞在Map通道处。而唯一的老化worker在任务结束后会面临两个结果：要么执行完任务后提交，系统切换到reduce phase，而其他worker持续阻塞；要么执行任务失败，MapTask会被并发的协程重新放回到MapTask通道中，随机一个worker获得任务，而后重复上述过程，此时老化的worker依旧被阻塞。

然而，此时带来的唯一结果是：只有单一worker执行reduce任务。这是我们不希望看到的。

这也是前文所述Wait阶段的用处所在。将Map阶段严格分成任务全部下发——任务全部完成两个阶段，对于任务全部下发的情况，worker不会阻塞在Map通道处，而是等待一段时间，重新申请任务。

二.worker.go

1.Map处理流程
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
//given mapf func & rpcreply, return the slice of intermediate data
func MapProcess(mapf func(string, string) []KeyValue, task *RPCReply) ([]string, bool) {
	filename := task.MapTask.Infile
	file, err := os.Open(filename)

	if(err != nil){
		log.Fatalf("cannot open %v", filename)
	}
	
	content, err := ioutil.ReadAll(file)


	if(err != nil){
		log.Fatalf("cannot read %v", filename)
	}
	
	file.Close()
	
	kvs := mapf(filename, string(content))//a slice of kv results
	
	//change kvs into tmpKvs
	nReducer := task.NReducer
	
	tmpKvs := make([][]KeyValue,nReducer)
	for _, kv := range kvs {
		index := ihash(kv.Key) % nReducer //index: results of hash equals to reduceNo
		tmpKvs[index] = append(tmpKvs[index], kv)
	}
	
	//write to files
	outFiles := make([]*os.File, nReducer)
	fileEncs := make([]*json.Encoder, nReducer)
	outFileNames := make([]string, nReducer)


	//write intermediate data to partitions with json
	for index, tmpKv := range tmpKvs {
		if len(tmpKv) > 0 {
			oname := "mr-" + strconv.Itoa(task.MapTask.No) + "-" + strconv.Itoa(index + 1)
			outFiles[index], err = os.Create(oname)
	
			if(err != nil) {
				fmt.Println("create intermediate wrong!")
				return outFileNames, false
			}
	
			fileEncs[index] = json.NewEncoder(outFiles[index])
	
			err := fileEncs[index].Encode(tmpKv)
	
			if(err != nil) {
				fmt.Println("write wrong!")
				return outFileNames, false
			}
	
			outFiles[index].Close()
			outFileNames[index] = oname
	
		}
	}
	
	return outFileNames, true
}

读入Map任务对应数据->整理成kv形式数组->将数组的所有内容以key即单词字符串为基准映射到相应区块->以json的形式保存为中间文件。由于相同的字符串会映射到同一个文件，因此无需担心计数错误的情况发生。

2.Reduce处理流程
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
func ReduceProcess(reducef func(string, []string) string, task *RPCReply) (string, bool) {
	filenames := task.ReduceTask.Infiles

	kvs := []KeyValue{}
	
	for _, filename :=  range filenames {
		file, err := os.Open(filename)
	
		if err != nil {
			continue
		}
	
		dec := json.NewDecoder(file)//decoder
	
		for {
			var tmp	Kv []KeyValue
			if err := dec.Decode(&tmpKv); err != nil {
				break
			}
			kvs = append(kvs, tmpKv...)
		}
	
		//delete
		if file != nil {
			_ = os.Remove(filename)
		//	file.Close()	//it's necessary because of test_case_4
		}
	}
	
	//sort, External sort is essential if the memory isn't enough. you can use heap
	sort.Sort(ByKey(kvs))
	
	oname := "mr-out-" + strconv.Itoa(task.ReduceTask.No)
	ofile, err := os.Create(oname)
	defer ofile.Close()
	
	if err != nil {
		fmt.Println("create file failed!")
		return oname, false
	}
	
	values := []string{}
	
	for i, kv := range kvs {
		values = append(values, kv.Value)
		//add the last one
		if i+1 < len(kvs) && kvs[i].Key == kvs[i+1].Key {
			continue
		}
	
		count := reducef(kv.Key, values)
		//infile : kv.key count
	
		ofile.WriteString(kv.Key + " " + count + "\n")
	
		//empty values
		values = values[0:0]
	}
	return oname, true

}

读入Reduce任务处理文件->排序->统计个数->输出为最终结果。如果排序时内存不足可以采用外部排序的方式。

测试结果

pic2

pic3

 

附录：mr包下代码全文

master.go

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
package mr

import (
	"fmt"
	"log"
	"sync"
	"time"
	"net"
	"os"
	"net/rpc"
	"net/http"
)


const MapPhase, ReducePhase, Wait, Finish = 0, 1, 2, 3

type Master struct {
	Phase int
	MappersChan chan MapTask
	ReducersChan chan ReduceTask

	NMapper int
	NReducer int
	
	FinishedMapper int
	FinishedReducer int
	
	MapOutPaths [][]string
	ReduceOutPaths []string
	
	mutex sync.Mutex

}

type MapTask struct {
	Infile string
	No int
}

type ReduceTask struct {
	Infiles []string
	No int
}

// Your code here -- RPC handlers for the worker to call.

//
// an example RPC handler.
//
// the RPC argument and reply types are defined in rpc.go.
//
func (m *Master) Example(args *ExampleArgs, reply *ExampleReply) error {
	reply.Y = args.X + 1
	return nil
}


func (m *Master) SendTask(args *RPCArgs, reply *RPCReply) error {

	//mutex the phase
	m.mutex.Lock()
	reply.Phase = m.Phase
	m.mutex.Unlock()
	
	switch reply.Phase {
	case MapPhase:
		reply.MapTask = <- m.MappersChan
		reply.NReducer = m.NReducer
		//avoid crash
		go m.MapTimeout(reply.MapTask)
	case ReducePhase:
		reply.ReduceTask = <- m.ReducersChan
		go m.ReduceTimeout(reply.ReduceTask)
	}
	
	return nil

}

func (m *Master) MapTimeout(mTask MapTask) {
	time.Sleep(time.Second * 10)

	if len(m.MapOutPaths[mTask.No - 1]) == 0 {
		fmt.Printf("Mapper No.%v crash!\n", mTask.No)
		m.Phase = MapPhase
		m.MappersChan <- mTask
	}

}

func (m *Master) ReduceTimeout(rTask ReduceTask) {
	time.Sleep(time.Second * 10)

	if len(m.ReduceOutPaths[rTask.No - 1]) == 0 {
		fmt.Printf("Reducer No.%v crash!\n", rTask.No)
		m.ReducersChan <- rTask
	}

}


func (m *Master) ReportTask(args *RPCArgs, reply *RPCReply) error {
	m.mutex.Lock()
	defer m.mutex.Unlock()
	switch args.Phase {
	case MapPhase:
		//to avoid repeating report

		if len(m.MapOutPaths[args.No - 1]) != 0 {
			return nil
		}
	
		//add single OutPaths -> MapOutPaths
		for i := range args.OutPaths {
			m.MapOutPaths[args.No - 1] = append(m.MapOutPaths[args.No - 1], args.OutPaths[i])
		}
		m.FinishedMapper++
		//MappersChan is empty
		if len(m.MappersChan) == 0 {
			m.Phase = Wait
		}
	
		//MapPhase over
		if m.NMapper == m.FinishedMapper {
			m.shuffle()
			m.Phase = ReducePhase
		}
	
	case ReducePhase:
	
		if len(m.ReduceOutPaths[args.No - 1]) != 0 {
			return nil
		}
	
		m.ReduceOutPaths[args.No - 1] = args.OutPath
		m.FinishedReducer++
		//ReducePhase over
		if m.NReducer == m.FinishedReducer {
			m.Phase = Finish
			fmt.Println("master finished.")
		}
	}
	
	return nil

}

//MapPhase -> ReducePhase
func (m *Master) shuffle() {
	for i := 0; i < m.NReducer; i++ {
		task := ReduceTask {
			No: i + 1,
		}
		for j := 0; j < m.NMapper; j++ {
			task.Infiles = append(task.Infiles, m.MapOutPaths[j][i])
		}
		m.ReducersChan <- task
	}
}

//
// start a thread that listens for RPCs from worker.go
//
func (m *Master) server() {
	rpc.Register(m)
	rpc.HandleHTTP()
	//l, e := net.Listen("tcp", ":1234")
	sockname := masterSock()
	os.Remove(sockname)
	l, e := net.Listen("unix", sockname)
	if e != nil {
		log.Fatal("listen error:", e)
	}
	go http.Serve(l, nil)
}

//
// main/mrmaster.go calls Done() periodically to find out
// if the entire job has finished.
//
func (m *Master) Done() bool {


	return m.Phase == Finish

}

//
// create a Master.
// main/mrmaster.go calls this function.
// nReduce is the number of reduce tasks to use.
//
func MakeMaster(files []string, nReducer int) *Master {
	m := Master{}
	m.NMapper = len(files)
	m.NReducer = nReducer

	m.MapOutPaths = make([][]string, m.NMapper)
	m.ReduceOutPaths = make([]string, m.NReducer)
	
	//create buffered channel
	m.MappersChan = make(chan MapTask,m.NMapper)
	m.ReducersChan = make(chan ReduceTask,m.NReducer)
	
	//MapperChan create
	for i, file := range files{
		mapper := MapTask{
			No : i + 1,
			Infile : file,
		}
		m.MappersChan <- mapper
	}
	m.Phase = MapPhase
	
	m.server()
	fmt.Println("master is working.")
	return &m

}

worker.go

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
package mr

import (
	"encoding/json"
	"fmt"
	"io/ioutil"
	"os"
	"sort"
	"strconv"
	"time"
    "log"
    "net/rpc"
    "hash/fnv"
)


//sort by key
type ByKey []KeyValue
func (b ByKey) Len() int 			{ return len(b) }
func (b ByKey) Swap(i, j int) 		{ b[i], b[j] = b[j], b[i]}
func (b ByKey) Less(i, j int) bool  { return b[i].Key < b[j].Key }

//
// Map functions return a slice of KeyValue.
//
type KeyValue struct {
	Key   string
	Value string
}

//
// use ihash(key) % NReduce to choose the reduce
// task number for each KeyValue emitted by Map.
//
func ihash(key string) int {
	h := fnv.New32a()
	h.Write([]byte(key))
	return int(h.Sum32() & 0x7fffffff)
}


//
// main/mrworker.go calls this function.
//
func Worker(mapf func(string, string) []KeyValue,
	reducef func(string, []string) string) {

	// Your worker implementation here.
	
	// uncomment to send the Example RPC to the master.
	
	// keep on handling data
	
	for{
		reply := AskTask()
		if reply == nil {
			fmt.Println("master finished, worker prepares to finish.")
			return
		}
		switch reply.Phase {
		case MapPhase:
			outPaths, ok := MapProcess(mapf, reply)
			if ok {
				//fmt.Printf("MapProcess %v OK.\n", reply.MapTask.No)
				call("Master.ReportTask", &RPCArgs{Phase: MapPhase, OutPaths: outPaths, No: reply.MapTask.No}, reply)
			} else {
				fmt.Printf("MapProcess %v failed.\n", reply.MapTask.No)
			}
		case ReducePhase:
			outPath, ok := ReduceProcess(reducef, reply)
			if ok {
				//fmt.Printf("ReduceProcess %v OK.\n", reply.ReduceTask.No)
				call("Master.ReportTask", &RPCArgs{Phase:ReducePhase, OutPath: outPath, No: reply.ReduceTask.No}, reply)
			} else {
				fmt.Printf("ReduceProcess %v failed.\n", reply.ReduceTask.No)
			}
		case Finish:
			fmt.Println("master finished, worker prepares to finish.")
			return
		case Wait:
			time.Sleep(time.Millisecond)
		}
	
	}

}

//given mapf func & rpcreply, return the slice of intermediate data
func MapProcess(mapf func(string, string) []KeyValue, task *RPCReply) ([]string, bool) {
	filename := task.MapTask.Infile
	file, err := os.Open(filename)

	if(err != nil){
		log.Fatalf("cannot open %v", filename)
	}
	
	content, err := ioutil.ReadAll(file)


	if(err != nil){
		log.Fatalf("cannot read %v", filename)
	}
	
	file.Close()
	
	kvs := mapf(filename, string(content))//a slice of kv results
	
	//change kvs into tmpKvs
	nReducer := task.NReducer
	
	tmpKvs := make([][]KeyValue,nReducer)
	for _, kv := range kvs {
		index := ihash(kv.Key) % nReducer //index: results of hash equals to reduceNo
		tmpKvs[index] = append(tmpKvs[index], kv)
	}
	
	//write to files
	outFiles := make([]*os.File, nReducer)
	fileEncs := make([]*json.Encoder, nReducer)
	outFileNames := make([]string, nReducer)


	//write intermediate data to partitions with json
	for index, tmpKv := range tmpKvs {
		if len(tmpKv) > 0 {
			oname := "mr-" + strconv.Itoa(task.MapTask.No) + "-" + strconv.Itoa(index + 1)
			outFiles[index], err = os.Create(oname)
	
			if(err != nil) {
				fmt.Println("create intermediate wrong!")
				return outFileNames, false
			}
	
			fileEncs[index] = json.NewEncoder(outFiles[index])
	
			err := fileEncs[index].Encode(tmpKv)
	
			if(err != nil) {
				fmt.Println("write wrong!")
				return outFileNames, false
			}
	
			outFiles[index].Close()
			outFileNames[index] = oname
	
		}
	}
	
	return outFileNames, true

}


func ReduceProcess(reducef func(string, []string) string, task *RPCReply) (string, bool) {
	filenames := task.ReduceTask.Infiles

	kvs := []KeyValue{}
	
	for _, filename :=  range filenames {
		file, err := os.Open(filename)
	
		if err != nil {
			continue
		}
	
		dec := json.NewDecoder(file)//decoder
	
		for {
			var tmpKv []KeyValue
			if err := dec.Decode(&tmpKv); err != nil {
				break
			}
			kvs = append(kvs, tmpKv...)
		}
	
		//delete
		if file != nil {
			_ = os.Remove(filename)
		//	file.Close()	//it's necessary because of test_case_4
		}
	}
	
	//sort, External sort is essential if the memory isn't enough. you can use heap
	sort.Sort(ByKey(kvs))
	
	oname := "mr-out-" + strconv.Itoa(task.ReduceTask.No)
	ofile, err := os.Create(oname)
	defer ofile.Close()
	
	if err != nil {
		fmt.Println("create file failed!")
		return oname, false
	}
	
	values := []string{}
	
	for i, kv := range kvs {
		values = append(values, kv.Value)
		//add the last one
		if i+1 < len(kvs) && kvs[i].Key == kvs[i+1].Key {
			continue
		}
	
		count := reducef(kv.Key, values)
		//infile : kv.key count
	
		ofile.WriteString(kv.Key + " " + count + "\n")
	
		//empty values
		values = values[0:0]
	}
	return oname, true

}
//ask for a task (map/reduce)
func AskTask() *RPCReply {
	args := &RPCArgs{}
	reply := &RPCReply{}
	err := call("Master.SendTask", args, reply)
	if !err {
		return nil
	}
	return reply
}
//
// example function to show how to make an RPC call to the master.
//
// the RPC argument and reply types are defined in rpc.go.
//
func CallExample() {

	// declare an argument structure.
	args := ExampleArgs{}
	
	// fill in the argument(s).
	args.X = 99
	
	// declare a reply structure.
	reply := ExampleReply{}
	
	// send the RPC request, wait for the reply.
	call("Master.Example", &args, &reply)
	
	// reply.Y should be 100.
	fmt.Printf("reply.Y %v\n", reply.Y)

}

//
// send an RPC request to the master, wait for the response.
// usually returns true.
// returns false if something goes wrong.
//
func call(rpcname string, args interface{}, reply interface{}) bool {
	// c, err := rpc.DialHTTP("tcp", "127.0.0.1"+":1234")
	sockname := masterSock()
	c, err := rpc.DialHTTP("unix", sockname)
	if err != nil {
		log.Fatal("dialing:", err)
	}
	defer c.Close()

	err = c.Call(rpcname, args, reply)
	if err == nil {
		return true
	}
	
	fmt.Println(err)
	return false

}

rpc.go

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
package mr

//
// RPC definitions.
//
// remember to capitalize all names.
//

import "os"
import "strconv"

//
// example to show how to declare the arguments
// and reply for an RPC.
//

type ExampleArgs struct {
	X int
}

type ExampleReply struct {
	Y int
}

// Add your RPC definitions here.
type  RPCArgs struct {
	//0: mapper 1: reducer
	Phase int
	No int
	OutPaths []string //OMapper
	OutPath string  //OReducer
}

type RPCReply struct {
	//0: mapper 1: reducer 2: wait 3: finish
	Phase int
	MapTask MapTask
	ReduceTask ReduceTask
	NReducer int
}


// Cook up a unique-ish UNIX-domain socket name
// in /var/tmp, for the master.
// Can't use the current directory since
// Athena AFS doesn't support UNIX-domain sockets.
func masterSock() string {
	s := "/var/tmp/824-mr-"
	s += strconv.Itoa(os.Getuid())
	return s
}
赏

谢谢你请我吃糖果

扫一扫，分享到微信

微信分享二维码

tag:

    缺失模块。
    1、请确保node版本大于6.2
    2、在博客根目录（注意不是yilia根目录）执行以下命令：
    npm i hexo-generator-json-content --save

    3、在根目录_config.yml里添加配置： 

      jsonContent:
    meta: false
    pages: false
    posts:
      title: true
      date: true
      path: true
      text: false
      raw: false
      content: false
      slug: false
      updated: false
      comments: false
      link: false
      permalink: false
      excerpt: false
      categories: false
      tags: true

热爱跳舞<br><br>永远是学徒