6.584-Lab1:MapReduce

news/2024/11/14 0:45:11/

前置知识/概念

Raft

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

RPC(Remote Producer Call)

参考链接1
参考链接2

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 intconst (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 TaskReduceAlloc                  // 表示coordinator向worker分配Reduce TaskWait                         // 表示coordinator让worker休眠Shutdown                     // 表示coordinator让worker终止
)

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

type MsgSend struct {MsgType MsgTypeTaskId  int
}

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

type MsgReply struct {MsgType  MsgTypeNReduce  intTaskId   int    // 当worker发送MsgSend申请任务、coordinator回复任务的IDTaskName 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 taskerr := 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.NReduceif encoders[redId] == nil {tempFile, err := ioutil.TempFile("", fmt.Sprintf("mr-map-tmp-%d", redId))if err != nil {return err}defer tempFile.Close()tempFiles[redId] = tempFileencoders[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) % NReducek_vs := map[string][]string{}for _, file := range files {dec := json.NewDecoder(file)for { // 循环读JSON数据流var kv KeyValue // 将读出的JSON数据解码放入kvif 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-TaskIdofile, 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]*MapTaskInfomu          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超时的任务分配给workerMapSuccessNum := 0 // Map task 完成个数for filename, maptaskinfo := range c.MapTasks {alloc := falseif 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.TaskIdreply.TaskName = filenamereply.NReduce = c.NReducereply.MsgType = MapAllocmaptaskinfo.statue = runningmaptaskinfo.startTime = time.Now().Unix()return nil}}// 没有任务可以分配但所有任务没有完成if MapSuccessNum < len(c.MapTasks) {reply.MsgType = Waitreturn nil}// 运行到这里表明所有的Map任务都已经完成ReduceSuccessNum := 0for idex, reducetaskinfo := range c.ReduceTasks {alloc := falseif 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 = idexreply.NReduce = c.NReducereply.MsgType = ReduceAllocreducetaskinfo.statue = runningreducetaskinfo.startTime = time.Now().Unix()return nil}}if ReduceSuccessNum < len(c.ReduceTasks) {reply.MsgType = Waitreturn nil}// 运行到这里表明所有的任务都已完成reply.MsgType = Shutdownreturn 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,否则返回falsefor _, 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.ppmy.cn/news/1546787.html

相关文章

⚙️ 如何调整重试策略以适应不同的业务需求?

调整 Kafka 生产者和消费者的重试策略以适应不同的业务需求&#xff0c;需要根据业务的特性和容错要求来进行细致的配置。以下是一些关键的调整策略&#xff1a; 业务重要性&#xff1a; 对于关键业务消息&#xff0c;可以增加重试次数&#xff0c;并设置较长的重试间隔&#x…

初识Linux · 消息队列和信号量

目录 前言&#xff1a; 消息队列 信号量 前言&#xff1a; 对于消息队列&#xff0c;信号量&#xff0c;共享内存都是隶属于system V这个标准下的进程间通信&#xff0c;其实上文的共享内存已经是基本上快被淘汰的了&#xff0c;对于其他的两个&#xff0c;消息队列和信号量…

C++开发基础之使用librabbitmq库实现RabbitMQ消息队列通信

1. 前言 RabbitMQ是一个流行的开源消息队列系统&#xff0c;支持多种消息协议&#xff0c;广泛用于构建分布式系统和微服务架构。可以在不同应用程序之间实现异步消息传递。在本文中&#xff0c;我们将熟悉如何使用C与RabbitMQ进行消息通信。 2. 准备工作 在 Windows 平台上…

计算机新手练级攻略——如何搜索问题

目录 计算机学生新手练级攻略——如何搜索问题1.明确搜索意图2.使用精确关键词3.使用专业引擎搜索4.利用好技术社区1. Stack Overflow2. GitHub3. IEEE Xplore4. DBLP 5.使用代码搜索工具1. GitHub 代码搜索2. Stack Overflow 代码搜索3. Papers with Code4. IEEE Xplore 6.查阅…

【STM32】GPIO口以及EXTI外部中断

在STM32微控制器中&#xff0c;GPIO&#xff08;通用输入输出&#xff09;口和EXTI&#xff08;外部中断&#xff09;是实现用户交互和事件响应的重要功能。本文将详细介绍STM32的GPIO口和EXTI外部中断的配置和使用方法&#xff0c;以及如何通过代码实现这些功能。 GPIO口简介…

创新体验触手可及 紫光展锐携手影目科技推出AI眼镜开放平台

近日&#xff0c;紫光展锐携手影目科技共同发布了搭载展锐W517芯片的影目X系列AI眼镜开放平台。这一产品的推出标志着双方在智能穿戴领域的深度协作&#xff0c;将紫光展锐的领先芯片技术与影目的产品创新相融合&#xff0c;合力打造全球智能眼镜市场的标杆产品。这一战略布局不…

vue3中的组件通信方式有哪些?

在vue3里&#xff0c;组件是一个非常重要的概念&#xff0c;项目中各个组件间的通信也是一个非常常见的需求&#xff0c;接下来我将为大家展示vue3组件有哪几种常见的通信方式。 一、props 适用场景&#xff1a;父子组件之间的通信 父传子&#xff1a; 父组件在子组件的标签…

HarmonyOS入门简介

&#x1f341; 作者&#xff1a;知识浅谈&#xff0c;CSDN签约讲师&博客专家&#xff0c;华为云云享专家&#xff0c;阿里云专家博主&#xff0c;InfoQ签约作者 &#x1f4cc; 擅长领域&#xff1a;全栈工程师、爬虫、ACM算法&#xff0c;大数据&#xff0c;深度学习 &…