国产探花免费观看_亚洲丰满少妇自慰呻吟_97日韩有码在线_资源在线日韩欧美_一区二区精品毛片,辰东完美世界有声小说,欢乐颂第一季,yy玄幻小说排行榜完本

首頁 > 編程 > Golang > 正文

golang如何實現mapreduce單進程版本詳解

2020-04-01 18:59:53
字體:
來源:轉載
供稿:網友

前言

  MapReduce作為hadoop的編程框架,是工程師最常接觸的部分,也是除去了網絡環境和集群配 置之外對整個Job執行效率影響很大的部分,所以很有必要深入了解整個過程。元旦放假的第一天,在家沒事干,用golang/292804.html">golang/196407.html">golang實現了一下mapreduce的單進程版本,github地址。處理對大文件統計最高頻的10個單詞,因為功能比較簡單,所以設計沒有解耦合。

  本文先對mapreduce大體概念進行介紹,然后結合代碼介紹一下,如果接下來幾天有空,我會實現一下分布式高可用的mapreduce版本。下面話不多說了,來一起看看詳細的介紹吧。

1. Mapreduce大體架構

golang,mapreduce,進程管理,mapreduce包

  上圖是論文中mapreduce的大體架構。總的來說Mapreduce的思想就是分治思想:對數據進行分片,然后用mapper進行處理,以key-value形式輸出中間文件;然后用reducer進行對mapper輸出的中間文件進行合并:將key一致的合到一塊,并輸出結果文件;如果有需要,采用Combiner進行最后的合并。

  歸納來說主要分為5部分:用戶程序、Master、Mapper、Reducer、Combiner(上圖未給出)。

  • 用戶程序。用戶程序主要對輸入數據進行分割,制定Mapper、Reducer、Combiner的代碼。
  • Master:中控系統。控制分發Mapper、Reduer的個數,比如生成m個進程處理Mapper,n個進程處理Reducer。其實對Master來說,Mapper和Reduer都屬于worker,只不過跑的程序不一樣,Mapper跑用戶輸入的map代碼,Reduer跑用戶輸入的reduce代碼。Master還作為管道負責中間路徑傳遞,比如將Mapper生成的中間文件傳遞給Reduer,將Reduer生成的結果文件返回,或者傳遞給Combiner(如果有需要的話)。由于Master是單點,性能瓶頸,所以可以做集群:主備模式或者分布式模式。可以用zookeeper進行選主,用一些消息中間件進行數據同步。Master還可以進行一些策略處理:比如某個Worker執行時間特別長,很有可能卡住了,對分配給該Worker的數據重新分配給別的Worker執行,當然需要對多份數據返回去重處理。
  • Mapper:負責將輸入數據切成key-value格式。Mapper處理完后,將中間文件的路徑告知Master,Master獲悉后傳遞給Reduer進行后續處理。如果Mapper未處理完,或者已經處理完但是Reduer未讀完其中間輸出文件,分配給該Mapper的輸入將重新被別的Mapper執行。
  • Reducer: 接受Master發送的Mapper輸出文件的消息,RPC讀取文件并處理,并輸出結果文件。n個Reduer將產生n個輸出文件。
  • Combiner: 做最后的歸并處理,通常不需要。

  總的來說,架構不復雜。組件間通信用啥都可以,比如RPC、HTTP或者私有協議等。

2. 實現代碼介紹

  該版本代碼實現了單機單進程版本,Mapper、Reducer和Combiner的實現用協程goroutine實現,通信采用channel。代碼寫的比較隨意,沒有解耦合。

  • 功能:統計給定文件中出現的最高頻的10個單詞
  • 輸入:大文件
  • 輸出:最高頻的10個單詞
  • 實現:5個Mapper協程、2個Reducer、1個Combiner。

  為了方便起見,Combiner對最高頻的10個單詞進行堆排序處理,按規范來說應該放在用戶程序處理。

  文件目錄如下,其中bin文件夾下的big_input_file.txt為輸入文件,可以調用generate下的main文件生成,caller文件為入口的用戶程序,master目錄下分別存放master、mapper、reducer、combiner代碼:

.├── README.md├── bin│ └── file-store│  └── big_input_file.txt└── src ├── caller │ └── main.go ├── generate │ └── main.go └── master  ├── combiner.go  ├── mapper.go  ├── master.go  └── reducer.go6 directories, 8 files 

2.1 caller

  用戶程序,讀入文件并按固定行數進行劃分;然后調用master.Handle進行處理。

package mainimport (  "os" "path" "path/filepath" "bufio" "strconv" "master" "github.com/vinllen/go-logger/logger")const (  LIMIT int = 10000 // the limit line of every file)func main() {  curDir, err := filepath.Abs(filepath.Dir(os.Args[0])) if err != nil {  logger.Error("Read path error: ", err.Error())  return } fileDir := path.Join(curDir, "file-store") _ = os.Mkdir(fileDir, os.ModePerm) // 1. read file filename := "big_input_file.txt" inputFile, err := os.Open(path.Join(fileDir, filename)) if err != nil {  logger.Error("Read inputFile error: ", err.Error())  return } defer inputFile.Close() // 2. split inputFile into several pieces that every piece hold 100,000 lines filePieceArr := []string{} scanner := bufio.NewScanner(inputFile) piece := 1Outter:  for {  outputFilename := "input_piece_" + strconv.Itoa(piece)  outputFilePos := path.Join(fileDir, outputFilename)  filePieceArr = append(filePieceArr, outputFilePos)  outputFile, err := os.Create(outputFilePos)  if err != nil {   logger.Error("Split inputFile error: ", err.Error())   continue  }  defer outputFile.Close()  for cnt := 0; cnt < LIMIT; cnt++ {   if !scanner.Scan() {    break Outter   }   _, err := outputFile.WriteString(scanner.Text() + "/n")   if err != nil {    logger.Error("Split inputFile writting error: ", err.Error())    return   }  }  piece++ } // 3. pass to master res := master.Handle(filePieceArr, fileDir) logger.Warn(res)}

2.2 master

  Master程序,依次生成Combiner、Reducer、Mapper,處理消息中轉,輸出最后結果。

package masterimport ( "github.com/vinllen/go-logger/logger")var (  MapChanIn chan MapInput // channel produced by master while consumed by mapper MapChanOut chan string // channel produced by mapper while consumed by master ReduceChanIn chan string // channel produced by master while consumed by reducer ReduceChanOut chan string // channel produced by reducer while consumed by master CombineChanIn chan string // channel produced by master while consumed by combiner CombineChanOut chan []Item // channel produced by combiner while consumed by master)func Handle(inputArr []string, fileDir string) []Item {  logger.Info("handle called") const(  mapperNumber int = 5  reducerNumber int = 2 ) MapChanIn = make(chan MapInput) MapChanOut = make(chan string) ReduceChanIn = make(chan string) ReduceChanOut = make(chan string) CombineChanIn = make(chan string) CombineChanOut = make(chan []Item) reduceJobNum := len(inputArr) combineJobNum := reducerNumber // start combiner go combiner() // start reducer for i := 1; i <= reducerNumber; i++ {  go reducer(i, fileDir) } // start mapper for i := 1; i <= mapperNumber; i++ {  go mapper(i, fileDir) } go func() {  for i, v := range(inputArr) {   MapChanIn <- MapInput{    Filename: v,    Nr: i + 1,   } // pass job to mapper  }  close(MapChanIn) // close map input channel when no more job }() var res []Itemoutter:  for {  select {   case v := <- MapChanOut:    go func() {     ReduceChanIn <- v     reduceJobNum--     if reduceJobNum <= 0 {      close(ReduceChanIn)     }    }()   case v := <- ReduceChanOut:    go func() {     CombineChanIn <- v     combineJobNum--     if combineJobNum <= 0 {      close(CombineChanIn)     }    }()   case v := <- CombineChanOut:    res = v    break outter  } } close(MapChanOut) close(ReduceChanOut) close(CombineChanOut) return res}

2.3 mapper

  Mapper程序,讀入并按key-value格式生成中間文件,告知Master。

package masterimport (  "fmt" "path" "os" "bufio" "strconv" "github.com/vinllen/go-logger/logger")type MapInput struct {  Filename string Nr int}func mapper(nr int, fileDir string) {  for {  val, ok := <- MapChanIn // val: filename  if !ok { // channel close   break  }  inputFilename := val.Filename  nr := val.Nr  file, err := os.Open(inputFilename)  if err != nil {   errMsg := fmt.Sprintf("Read file(%s) error in mapper(%d)", inputFilename, nr)   logger.Error(errMsg)   MapChanOut <- ""   continue  }  mp := make(map[string]int)  scanner := bufio.NewScanner(file)  scanner.Split(bufio.ScanWords)  for scanner.Scan() {   str := scanner.Text()   //logger.Info(str)   mp[str]++  }  outputFilename := path.Join(fileDir, "mapper-output-" + strconv.Itoa(nr))  outputFileHandler, err := os.Create(outputFilename)  if err != nil {   errMsg := fmt.Sprintf("Write file(%s) error in mapper(%d)", outputFilename, nr)   logger.Error(errMsg)  } else {   for k, v := range mp {    str := fmt.Sprintf("%s %d/n", k, v)    outputFileHandler.WriteString(str)   }   outputFileHandler.Close()  }  MapChanOut <- outputFilename }}

2.4 reducer

  Reducer程序,讀入Master傳遞過來的中間文件并歸并。

package masterimport (  "fmt" "bufio" "os" "strconv" "path" "strings" "github.com/vinllen/go-logger/logger")func reducer(nr int, fileDir string) {  mp := make(map[string]int) // store the frequence of words // read file and do reduce for {  val, ok := <- ReduceChanIn  if !ok {   break  }  logger.Debug("reducer called: ", nr)  file, err := os.Open(val)  if err != nil {   errMsg := fmt.Sprintf("Read file(%s) error in reducer", val)   logger.Error(errMsg)   continue  }  scanner := bufio.NewScanner(file)  for scanner.Scan() {   str := scanner.Text()   arr := strings.Split(str, " ")   if len(arr) != 2 {    errMsg := fmt.Sprintf("Read file(%s) error that len of line(%s) != 2(%d) in reducer", val, str, len(arr))    logger.Warn(errMsg)    continue   }   v, err := strconv.Atoi(arr[1])   if err != nil {    errMsg := fmt.Sprintf("Read file(%s) error that line(%s) parse error in reduer", val, str)    logger.Warn(errMsg)    continue   }   mp[arr[0]] += v  }  if err := scanner.Err(); err != nil {   logger.Error("reducer: reading standard input:", err)  }  file.Close() } outputFilename := path.Join(fileDir, "reduce-output-" + strconv.Itoa(nr)) outputFileHandler, err := os.Create(outputFilename) if err != nil {  errMsg := fmt.Sprintf("Write file(%s) error in reducer(%d)", outputFilename, nr)  logger.Error(errMsg) } else {  for k, v := range mp {   str := fmt.Sprintf("%s %d/n", k, v)   outputFileHandler.WriteString(str)  }  outputFileHandler.Close() } ReduceChanOut <- outputFilename}

2.5 combiner

  Combiner程序,讀入Master傳遞過來的Reducer結果文件并歸并成一個,然后堆排序輸出最高頻的10個詞語。

package masterimport (  "fmt" "strings" "bufio" "os" "container/heap" "strconv" "github.com/vinllen/go-logger/logger")type Item struct {  key string val int}type PriorityQueue []*Itemfunc (pq PriorityQueue) Len() int {  return len(pq)}func (pq PriorityQueue) Less(i, j int) bool {  return pq[i].val > pq[j].val}func (pq PriorityQueue) Swap(i, j int) {  pq[i], pq[j] = pq[j], pq[i]}func (pq *PriorityQueue) Push(x interface{}) {  item := x.(*Item) *pq = append(*pq, item)}func (pq *PriorityQueue) Pop() interface{} {  old := *pq n := len(old) item := old[n - 1] *pq = old[0 : n - 1] return item}func combiner() {  mp := make(map[string]int) // store the frequence of words // read file and do combine for {  val, ok := <- CombineChanIn  if !ok {   break  }  logger.Debug("combiner called")  file, err := os.Open(val)  if err != nil {   errMsg := fmt.Sprintf("Read file(%s) error in combiner", val)   logger.Error(errMsg)   continue  }  scanner := bufio.NewScanner(file)  for scanner.Scan() {   str := scanner.Text()   arr := strings.Split(str, " ")   if len(arr) != 2 {    errMsg := fmt.Sprintf("Read file(%s) error that len of line != 2(%s) in combiner", val, str)    logger.Warn(errMsg)    continue   }   v, err := strconv.Atoi(arr[1])   if err != nil {    errMsg := fmt.Sprintf("Read file(%s) error that line(%s) parse error in combiner", val, str)    logger.Warn(errMsg)    continue   }   mp[arr[0]] += v  }  file.Close() } // heap sort // pq := make(PriorityQueue, len(mp)) pq := make(PriorityQueue, 0) heap.Init(&pq) for k, v := range mp {  node := &Item {   key: k,   val: v,  }  // logger.Debug(k, v)  heap.Push(&pq, node) } res := []Item{} for i := 0; i < 10 && pq.Len() > 0; i++ {  node := heap.Pop(&pq).(*Item)  res = append(res, *node) } CombineChanOut <- res}

3. 總結

  不足以及未實現之處:

  • 各模塊間耦合性高
  • master單點故障未擴展
  • 未采用多進程實現,進程間采用RPC通信
  • 未實現單個Workder時間過長,另起Worker執行任務的代碼。

  接下來要是有空,我會實現分布式高可用的代碼,模塊間采用RPC通訊。

好了,以上就是這篇文章的全部內容了,希望本文的內容對大家的學習或者工作具有一定的參考學習價值,如果有疑問大家可以留言交流,謝謝大家對VEVB武林網的支持。


發表評論 共有條評論
用戶名: 密碼:
驗證碼: 匿名發表
主站蜘蛛池模板: 同江市| 曲周县| 荔波县| 祥云县| 平原县| 尼玛县| 古丈县| 秦安县| 韩城市| 安达市| 文水县| 雅江县| 连江县| 田东县| 双城市| 于都县| 凤山市| 伽师县| 伊通| 和静县| 耿马| 甘谷县| 老河口市| 开封市| 盐池县| 汝州市| 普陀区| 崇明县| 北海市| 和静县| 华池县| 元阳县| 手机| 鄂温| 龙川县| 突泉县| 晋江市| 石楼县| 鄂托克旗| 平原县| 旌德县|