当前位置： 首页 > article >正文

6.584-Lab1：MapReduce

article 2024/12/25 1:36:50

前置知识/概念

Raft

是一个基于“Leader”的协议，能够保证分布式网路的一致性。

RPC（Remote Producer Call）

参考链接1
参考链接2
Go中RPC的简单实现

Golang中regexp正则表达式的用法

https://gukaifeng.cn/posts/golang-zheng-ze-biao-da-shi-regexp-de-ji-ben-yong-fa/index.html

Golang中自定义类型Sort

在提供的排序方法sort.Ints、sort.Floats、sort.Strings的底层都分别实现了三个函数Len()、Less()、Swap()，所以在我们实现自定义类型排序的时候要实现上述三个函数。

实现

参考链接

概览

在这里插入图片描述
如图，Lab1要实现两个部分分别是Map（映射）和Reduce（规约），Master负责调度，分配任务在代码中是Coordinator，而Worker则负责具体的map task和reduce task。
Map Task具体是统计文件中的单词生成对应的键值对[key, value]，通过一个哈希函数Ihash将单词映射为key，存在中间文件，例如File1中有单词a、b对应键值对为[a, 1]，[b, 1]，分别存放在mr-out-1-ihash(a/b)%NReduce的中间文件中，其中NReduce是Reduce Task的个数，代码中为10。
Reduce Task具体是将中间文件mr-out-*-taskid中的键值对放入最终的文件mr-out-taskid中。

代码

rpc.go

Coordinator和Worker通过RPC进行通信，在文件rpc.go中需要设计相应的数据结构让其进行通信。分析可能的具体行为：Worker需要向Coordinator申请任务、Coordinator需要向Worker分配任务（map task & reduce task）、Worker向Coordinator回复任务的执行情况（成功、失败）、没有闲置任务分配时Coordinator告诉Worker等待（Wait）、所有任务完成告诉Worker结束（Shutdown）。
根据上述分析设计如下数据类型：

// 用不同数字表示不同信息的类别
type MsgType int

const (
	AskForTask    MsgType = iota // 表示worker向coordinator申请任务
	MapSucceed                   // 表示worker向coordinator传递Map Task完成
	MapFailed                    // 表示worker向coordinator传递Map Task失败
	ReduceSucceed                // 表示worker向coordinator传递Reduce Task完成
	ReduceFailed                 // 表示worker向coordinator传递Reduce Task失败
	MapAlloc                     // 表示coordinator向worker分配Map Task
	ReduceAlloc                  // 表示coordinator向worker分配Reduce Task
	Wait                         // 表示coordinator让worker休眠
	Shutdown                     // 表示coordinator让worker终止
)

从Worker视角出发，设计发送给Coordinator的信息结构体MsgSend，需要包含任务id以及任务执行情况：

type MsgSend struct {
	MsgType MsgType
	TaskId  int
}

从Coordinator视角出发，设计发送给Worker的信息的结构体MsgSend，需要包含任务的id即要处理的第几个文件用于中间文件的命名、任务类型、要处理的filename、NRduce：

type MsgReply struct {
	MsgType  MsgType
	NReduce  int
	TaskId   int    // 当worker发送MsgSend申请任务、coordinator回复任务的ID
	TaskName string //
}

worker.go

上面分析可知Worker的行为有：申请任务、执行任务、汇报任务执行情况。

// worker的任务就是不断请求任务、执行任务、报告执行状态
// main/mrworker.go calls this function.
func Worker(mapf func(string, string) []KeyValue,
	reducef func(string, []string) string) {

	// Your worker implementation here.
	for {
		// 不断请求
		replyMsg := CallForTask()
		switch replyMsg.MsgType {
		case MapAlloc: // coordinator分配了map task
			err := HandleMapTask(replyMsg, mapf)
			if err != nil { // Map Task任务完成
				_ = CallForReportStatus(MapFailed, replyMsg.TaskId)
			} else { // Map Task 任务失败
				_ = CallForReportStatus(MapSucceed, replyMsg.TaskId)
			}
		case ReduceAlloc:
			err := HandleReduceTask(replyMsg, reducef)
			if err != nil { // Map Task任务完成
				_ = CallForReportStatus(ReduceFailed, replyMsg.TaskId)
			} else { // Map Task 任务失败
				_ = CallForReportStatus(ReduceSucceed, replyMsg.TaskId)
			}
		case Wait:
			time.Sleep(time.Second * 10)
		case Shutdown:
			os.Exit(0)
		}
		time.Sleep(time.Second)
	}
}

其中申请任务、回复任务执行情况的函数均利用了rpc中的Call来实现向Coordinator进行通信。
在这里插入图片描述
在这里插入图片描述

其中执行Map Task 的执行函数HandleMapTask：

func HandleMapTask(reply *MessageReply, mapf func(string, string) []KeyValue) error {
	file, err := os.Open(reply.TaskName)
	if err != nil {
		return err
	}
	defer file.Close()

	content, err := io.ReadAll(file)
	if err != nil {
		return err
	}

	kva := mapf(reply.TaskName, string(content))
	sort.Sort(ByKey(kva))

	tempFiles := make([]*os.File, reply.NReduce)
	encoders := make([]*json.Encoder, reply.NReduce)

	for _, kv := range kva {
		redId := ihash(kv.Key) % reply.NReduce
		if encoders[redId] == nil {
			tempFile, err := ioutil.TempFile("", fmt.Sprintf("mr-map-tmp-%d", redId))
			if err != nil {
				return err
			}
			defer tempFile.Close()
			tempFiles[redId] = tempFile
			encoders[redId] = json.NewEncoder(tempFile)
		}
		err := encoders[redId].Encode(&kv)
		if err != nil {
			return err
		}
	}

	for i, file := range tempFiles {
		if file != nil {
			fileName := file.Name()
			file.Close()
			newName := fmt.Sprintf("mr-out-%d-%d", reply.TaskID, i)
			if err := os.Rename(fileName, newName); err != nil {
				return err
			}
		}
	}
	return nil
}

在这里插入图片描述
由函数TemFile的描述可知，该函数在创建时会按照pattern+随机字符串作为临时文件名字，即是不同程序同时调用该函数会创建不同的临时文件所以不存在资源竞争是并发安全的。
先将Map Task映射的键值对存入临时文件中，等全部放入临时文件后再将临时文件重命名为需要的中间文件的名字，因为重命名操作时原子性的。
如果不采用临时文件直接存入目标中间文件的话，会出现存入中间文件之前中间文件中含有其他数据即脏数据，可能是上个Worker执行到一半因为某些原因而退出之前存入的，所以存入之前需要将中间文件清空一下，这样就比较浪费时间。

执行Reduce Task 的执行函数HandleReduceTask也是同样的问题，虽然从中间件读取的时候没有写入操作但写入最终文件时也同样需要像上面一样保证原子性，这里贴出没有使用临时文件的代码：

// 处理分配的 Reduce 任务，处理每个MapTask产生的mr-out-*-key_id
func HandleReduceTask(reply *MsgReply, reducef func(string, []string) string) error {
	key_id := reply.TaskId
	// todo:这里的key_id要 % NReduce吗 --答：传入的TaskId一定是小于NReduce的，规约的数量就是NRduce也即Reduce Task的数量
	files, err := ReadSpecificFile(key_id, "./")
	if err != nil {
		return err
	}
	// 从所有匹配的文件中读出Json格式的键值对[k1-value]/[k2-value]，key哈希的值可以不一样但这里ihash(k1) % NReduce  = ihash(k2) % NReduce
	k_vs := map[string][]string{}
	for _, file := range files {
		dec := json.NewDecoder(file)
		for { // 循环读JSON数据流
			var kv KeyValue // 将读出的JSON数据解码放入kv
			if err := dec.Decode(&kv); err != nil {
				break
			}
			k_vs[kv.Key] = append(k_vs[kv.Key], kv.Value)
		}
		file.Close()
	}

	keys := []string{} // 将map中的keys拿出排序，按照keys的字典序依次写入文件
	for k, _ := range k_vs {
		keys = append(keys, k)
	}
	sort.Strings(keys)
	oname := "mr-out-" + strconv.Itoa(reply.TaskId) // 将Reduce后的结果放入 mr-out-TaskId
	ofile, err := os.Create(oname)
	if err != nil {
		return err
	}
	defer ofile.Close()
	for _, key := range keys {
		output := reducef(key, k_vs[key])
		_, err := fmt.Fprintf(ofile, "%s %s\n", key, output) // 格式化写入文件
		if err != nil {
			return err
		}
	}

	CleanFileByReduceId(reply.TaskId, "./")
	return nil
}

利用临时文件保证原子性参考上面HandleMapTask函数的实现。其中需要注意的是最终写入文件的时候Key要按照字典升序排列，而map存储的Key不是有序的，所以把Key拿出来排序再放入文件。

coordinator.go

coordinator中要实现worker申请任务的函数AskForTask以及对worker报告任务执行情况后对相应任务状态的更新函数NoticeResult。
任务的状态有闲置idle、成功finished、失败failed、超时、正在运行running。每次worker申请任务时都轮询一下所有任务，委派闲置、失败、超时的任务，而超时状态的判断则通过为每个任务打上运行开始的时间戳，若轮询到running的任务时判断当前时间与开始的时间戳比较若大于10s则可以判定为超时，可以再次委派给worker。

TaskInfo结构
type TaskStatue int

// 定义类别
const (
	idle     TaskStatue = iota // 闲置
	finished                   // 完成
	running                    // 运行
	failed                     // 失败
)

type MapTaskInfo struct {
	statue    TaskStatue // 任务状态
	TaskId    int        // 任务编号
	startTime int64      // 分配时间，当前时间-分配时间>10s表示超时
}
type ReduceTaskInfo struct {
	statue TaskStatue
	//TaskId reduce task的编号用数组下标表示
	startTime int64
}
type Coordinator struct {
	NReduce     int // reduce tasks可用的数量
	MapTasks    map[string]*MapTaskInfo
	mu          sync.Mutex // 互斥锁
	ReduceTasks []*ReduceTaskInfo
}

初始化函数：

// 初始化函数
func (c *Coordinator) Init(files []string) {
	for idx, filename := range files {
		c.MapTasks[filename] = &MapTaskInfo{
			statue: idle, // 初始为闲置
			TaskId: idx,
		}
	}
	for idx := 0; idx < c.NReduce; idx++ {
		c.ReduceTasks[idx] = &ReduceTaskInfo{
			statue: idle,
		}
	}
}

// create a Coordinator.
// main/mrcoordinator.go calls this function.
// nReduce is the number of reduce tasks to use.
func MakeCoordinator(files []string, nReduce int) *Coordinator {
	c := Coordinator{
		NReduce:     nReduce,
		MapTasks:    make(map[string]*MapTaskInfo),
		ReduceTasks: make([]*ReduceTaskInfo, nReduce),
	}
	c.Init(files)

	c.server()
	return &c
}
rpc相应函数AskForTask的实现

AskForTask：实现相对复杂，大致流程为：
1.每个任务初始化为闲置。
2.每当有worker申请任务的话就轮询所有任务。
3.若存在有idle、failed、超时的任务就可以分配。
4.若没有可分配的任务，判断完成任务的个数是否等于所有任务个数：
4.1 若相等，则表明所有任务完成，告知workershutdown。
4.2 若不相等，则表明有任务还在进行且没有可分配的任务，告知workerWait。

期间应保证共享资源的互斥，每个worker请求任务的同时要上锁。

func (c *Coordinator) AskForTask(req *MsgSend, reply *MsgReply) error {
	if req.MsgType != AskForTask { // 传入的不是“申请任务”类型的信息
		return NoMathMsgType
	}

	// 加锁，保证每个worker申请任务时互斥
	c.mu.Lock()
	defer c.mu.Unlock()
	// 选择一个失败or闲置or超时的任务分配给worker
	MapSuccessNum := 0 // Map task 完成个数
	for filename, maptaskinfo := range c.MapTasks {
		alloc := false
		if maptaskinfo.statue == idle || maptaskinfo.statue == failed { // 该任务闲置或失败则可以分配
			alloc = true
		} else if maptaskinfo.statue == running { // 判断该任务是否超时，若超时则再分配
			if time.Now().Unix()-maptaskinfo.startTime > 10 {
				maptaskinfo.startTime = time.Now().Unix() // 再分配更新开始时间
				alloc = true
			}
		} else { // 该任务是已完成任务
			MapSuccessNum++
		}

		// 当前任务可以分配
		if alloc {
			reply.TaskId = maptaskinfo.TaskId
			reply.TaskName = filename
			reply.NReduce = c.NReduce
			reply.MsgType = MapAlloc

			maptaskinfo.statue = running
			maptaskinfo.startTime = time.Now().Unix()
			return nil
		}
	}

	// 没有任务可以分配但所有任务没有完成
	if MapSuccessNum < len(c.MapTasks) {
		reply.MsgType = Wait
		return nil
	}

	// 运行到这里表明所有的Map任务都已经完成
	ReduceSuccessNum := 0
	for idex, reducetaskinfo := range c.ReduceTasks {
		alloc := false
		if reducetaskinfo.statue == idle || reducetaskinfo.statue == failed {
			alloc = true
		} else if reducetaskinfo.statue == running {
			if time.Now().Unix()-reducetaskinfo.startTime > 10 {
				reducetaskinfo.startTime = time.Now().Unix()
				alloc = true
			}
		} else {
			ReduceSuccessNum++
		}

		if alloc {
			reply.TaskId = idex
			reply.NReduce = c.NReduce
			reply.MsgType = ReduceAlloc

			reducetaskinfo.statue = running
			reducetaskinfo.startTime = time.Now().Unix()
			return nil
		}
	}
	if ReduceSuccessNum < len(c.ReduceTasks) {
		reply.MsgType = Wait
		return nil
	}

	// 运行到这里表明所有的任务都已完成
	reply.MsgType = Shutdown

	return nil
}
rpc相应函数NoticeResult的实现

只需要将worker传递过来的任务完成状态更新到coordinator的TaskInfo即可。

func (c *Coordinator) NoticeResult(req *MsgSend, reply *MsgReply) error {
	c.mu.Lock()
	defer c.mu.Unlock()
	if req.MsgType == MapSucceed {
		for _, taskinfo := range c.MapTasks {
			if taskinfo.TaskId == req.TaskId {
				taskinfo.statue = finished
			}
		}
	} else if req.MsgType == ReduceSucceed {
		c.ReduceTasks[req.TaskId].statue = finished
	} else if req.MsgType == MapFailed {
		for _, taskinfo := range c.MapTasks {
			if taskinfo.TaskId == req.TaskId {
				taskinfo.statue = failed
			}
		}
	} else if req.MsgType == ReduceFailed {
		c.ReduceTasks[req.TaskId].statue = failed
	}
	return nil
}
所有任务是否完成-Done函数

只需要轮询一遍任务数组，判断是否所有任务均完成。

// if the entire job has finished.
func (c *Coordinator) Done() bool {

	// Your code here.
	// 遍历所有任务，全部完成则返回true，否则返回false
	for _, taskinfo := range c.MapTasks {
		if taskinfo.statue != finished {
			return false
		}
	}

	for _, taskinfo := range c.ReduceTasks {
		if taskinfo.statue != finished {
			return false
		}
	}
	return true
}

http://www.kler.cn/a/394420.html

相关文章：

  • Vue 92 ，Element 15 ，Vue + el-upload 实现图片上传与管理
  • Redis篇--常见问题篇6--缓存一致性1（Mysql和Redis缓存一致，更新数据库删除缓存策略）
  • 0009.基于springboot+layui的ERP企业进销存管理系统
  • Java 优化springboot jar 内存 年轻代和老年代的比例 减少垃圾清理耗时 如调整 -XX:NewRatio
  • kotlin中泛型中in和out的区别
  • 知识图谱+RAG学习
  • 深入解析 OpenHarmony 构建系统-1
  • 制作图片木马
  • 为什么海外服务器IP会被封
  • WebKit（适用2024年11月份版本）
  • 笔记--（网络服务4）、远程访问及控制
  • JDBC使用方式（项目由于一些不可逆因素，必须要使用JDBC连接）
  • gcd的递归与非递归实现
  • opencv视频读写
  • 机器学习(1)
  • Substance Painter技巧及心得
  • 自動換IP為什麼會不穩定？
  • shell命令笔记
  • gitlab修改root密码详细详情,高版本通用
  • 35数据库服务器（如MySQL, PostgreSQL）
  • Puppeteer教程：使用CSS选择器点击和爬取动态数据
  • 手机版产品目录如何制作？
  • PdServer：调用MidjourneyAPI完成静夜思图文生成
  • PySpark——Python与大数据
  • 极狐GitLab 发布安全补丁版本17.5.2, 17.4.4, 17.3.7
  • 为什么在Ubuntu下使用VScode开发C++程序时需要手动配置链接库