小編給大家分享一下golang怎么實(shí)現(xiàn)mapreduce單進(jìn)程，相信大部分人都還不怎么了解，因此分享這篇文章給大家參考一下，希望大家閱讀完這篇文章后大有收獲，下面讓我們一起去了解一下吧！

創(chuàng)新互聯(lián)于2013年開(kāi)始，是專(zhuān)業(yè)互聯(lián)網(wǎng)技術(shù)服務(wù)公司，擁有項(xiàng)目成都做網(wǎng)站、網(wǎng)站制作網(wǎng)站策劃，項(xiàng)目實(shí)施與項(xiàng)目整合能力。我們以讓每一個(gè)夢(mèng)想脫穎而出為使命，1280元吳興做網(wǎng)站,已為上家服務(wù),為吳興各地企業(yè)和個(gè)人服務(wù),聯(lián)系電話(huà):13518219792

1. Mapreduce大體架構(gòu)

??上圖是論文中mapreduce的大體架構(gòu)。總的來(lái)說(shuō)Mapreduce的思想就是分治思想：對(duì)數(shù)據(jù)進(jìn)行分片，然后用mapper進(jìn)行處理，以key-value形式輸出中間文件；然后用reducer進(jìn)行對(duì)mapper輸出的中間文件進(jìn)行合并：將key一致的合到一塊，并輸出結(jié)果文件；如果有需要，采用Combiner進(jìn)行最后的合并。

??歸納來(lái)說(shuō)主要分為5部分：用戶(hù)程序、Master、Mapper、Reducer、Combiner（上圖未給出）。

• 用戶(hù)程序。用戶(hù)程序主要對(duì)輸入數(shù)據(jù)進(jìn)行分割，制定Mapper、Reducer、Combiner的代碼。

• Master：中控系統(tǒng)。控制分發(fā)Mapper、Reduer的個(gè)數(shù)，比如生成m個(gè)進(jìn)程處理Mapper，n個(gè)進(jìn)程處理Reducer。其實(shí)對(duì)Master來(lái)說(shuō)，Mapper和Reduer都屬于worker，只不過(guò)跑的程序不一樣，Mapper跑用戶(hù)輸入的map代碼，Reduer跑用戶(hù)輸入的reduce代碼。Master還作為管道負(fù)責(zé)中間路徑傳遞，比如將Mapper生成的中間文件傳遞給Reduer，將Reduer生成的結(jié)果文件返回，或者傳遞給Combiner（如果有需要的話(huà)）。由于Master是單點(diǎn)，性能瓶頸，所以可以做集群：主備模式或者分布式模式。可以用zookeeper進(jìn)行選主，用一些消息中間件進(jìn)行數(shù)據(jù)同步。Master還可以進(jìn)行一些策略處理：比如某個(gè)Worker執(zhí)行時(shí)間特別長(zhǎng)，很有可能卡住了，對(duì)分配給該Worker的數(shù)據(jù)重新分配給別的Worker執(zhí)行，當(dāng)然需要對(duì)多份數(shù)據(jù)返回去重處理。

• Mapper：負(fù)責(zé)將輸入數(shù)據(jù)切成key-value格式。Mapper處理完后，將中間文件的路徑告知Master，Master獲悉后傳遞給Reduer進(jìn)行后續(xù)處理。如果Mapper未處理完，或者已經(jīng)處理完但是Reduer未讀完其中間輸出文件，分配給該Mapper的輸入將重新被別的Mapper執(zhí)行。

• Reducer: 接受Master發(fā)送的Mapper輸出文件的消息，RPC讀取文件并處理，并輸出結(jié)果文件。n個(gè)Reduer將產(chǎn)生n個(gè)輸出文件。

• Combiner: 做最后的歸并處理，通常不需要。

??總的來(lái)說(shuō)，架構(gòu)不復(fù)雜。組件間通信用啥都可以，比如RPC、HTTP或者私有協(xié)議等。

2. 實(shí)現(xiàn)代碼介紹

??該版本代碼實(shí)現(xiàn)了單機(jī)單進(jìn)程版本，Mapper、Reducer和Combiner的實(shí)現(xiàn)用協(xié)程goroutine實(shí)現(xiàn)，通信采用channel。代碼寫(xiě)的比較隨意，沒(méi)有解耦合。

• 功能：統(tǒng)計(jì)給定文件中出現(xiàn)的最高頻的10個(gè)單詞

• 輸入：大文件

• 輸出：最高頻的10個(gè)單詞

• 實(shí)現(xiàn)：5個(gè)Mapper協(xié)程、2個(gè)Reducer、1個(gè)Combiner。

??為了方便起見(jiàn)，Combiner對(duì)最高頻的10個(gè)單詞進(jìn)行堆排序處理，按規(guī)范來(lái)說(shuō)應(yīng)該放在用戶(hù)程序處理。

??文件目錄如下，其中bin文件夾下的big_input_file.txt為輸入文件，可以調(diào)用generate下的main文件生成，caller文件為入口的用戶(hù)程序，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.go

6 directories, 8 files

2.1 caller

??用戶(hù)程序，讀入文件并按固定行數(shù)進(jìn)行劃分；然后調(diào)用master.Handle進(jìn)行處理。

package main
import ( 
 "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 := 1
Outter: 
 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，處理消息中轉(zhuǎn)，輸出最后結(jié)果。

package master
import (
 "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 []Item
outter: 
 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 master
import ( 
 "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傳遞過(guò)來(lái)的中間文件并歸并。

package master
import ( 
 "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傳遞過(guò)來(lái)的Reducer結(jié)果文件并歸并成一個(gè)，然后堆排序輸出最高頻的10個(gè)詞語(yǔ)。

package master
import ( 
 "fmt"
 "strings"
 "bufio"
 "os"
 "container/heap"
 "strconv"

 "github.com/vinllen/go-logger/logger"
)
type Item struct { 
 key string
 val int
}
type PriorityQueue []*Item
func (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
}

以上是“golang怎么實(shí)現(xiàn)mapreduce單進(jìn)程”這篇文章的所有內(nèi)容，感謝各位的閱讀！相信大家都有了一定的了解，希望分享的內(nèi)容對(duì)大家有所幫助，如果還想學(xué)習(xí)更多知識(shí)，歡迎關(guān)注創(chuàng)新互聯(lián)成都網(wǎng)站設(shè)計(jì)公司行業(yè)資訊頻道！

另外有需要云服務(wù)器可以了解下創(chuàng)新互聯(lián)scvps.cn，海內(nèi)外云服務(wù)器15元起步，三天無(wú)理由+7*72小時(shí)售后在線(xiàn)，公司持有idc許可證，提供“云服務(wù)器、裸金屬服務(wù)器、高防服務(wù)器、香港服務(wù)器、美國(guó)服務(wù)器、虛擬主機(jī)、免備案服務(wù)器”等云主機(jī)租用服務(wù)以及企業(yè)上云的綜合解決方案，具有“安全穩(wěn)定、簡(jiǎn)單易用、服務(wù)可用性高、性?xún)r(jià)比高”等特點(diǎn)與優(yōu)勢(shì)，專(zhuān)為企業(yè)上云打造定制，能夠滿(mǎn)足用戶(hù)豐富、多元化的應(yīng)用場(chǎng)景需求。

本文題目：golang怎么實(shí)現(xiàn)mapreduce單進(jìn)程-創(chuàng)新互聯(lián)
鏈接分享：http://www.chinadenli.net/article20/hssco.html

成都網(wǎng)站建設(shè)公司_創(chuàng)新互聯(lián)，為您提供品牌網(wǎng)站建設(shè)、網(wǎng)站建設(shè)、標(biāo)簽優(yōu)化、面包屑導(dǎo)航、微信小程序、ChatGPT

廣告

聲明：本網(wǎng)站發(fā)布的內(nèi)容（圖片、視頻和文字）以用戶(hù)投稿、用戶(hù)轉(zhuǎn)載內(nèi)容為主，如果涉及侵權(quán)請(qǐng)盡快告知，我們將會(huì)在第一時(shí)間刪除。文章觀點(diǎn)不代表本網(wǎng)站立場(chǎng)，如需處理請(qǐng)聯(lián)系客服。電話(huà)：028-86922220；郵箱：631063699@qq.com。內(nèi)容未經(jīng)允許不得轉(zhuǎn)載，或轉(zhuǎn)載時(shí)需注明來(lái)源： 創(chuàng)新互聯(lián)