前言
MapReduce作為hadoop的編程框架,是工程師最常接觸的部分,也是除去了網絡環境和集群配 置之外對整個Job執行效率影響很大的部分,所以很有必要深入了解整個過程。元旦放假的第一天，在家沒事干，用golang/292804.html">golang/196407.html">golang實現了一下mapreduce的單進程版本，github地址。處理對大文件統計最高頻的10個單詞，因為功能比較簡單，所以設計沒有解耦合。
本文先對mapreduce大體概念進行介紹，然后結合代碼介紹一下，如果接下來幾天有空，我會實現一下分布式高可用的mapreduce版本。下面話不多說了，來一起看看詳細的介紹吧。
1. Mapreduce大體架構
上圖是論文中mapreduce的大體架構。總的來說Mapreduce的思想就是分治思想：對數據進行分片，然后用mapper進行處理，以key-value形式輸出中間文件；然后用reducer進行對mapper輸出的中間文件進行合并：將key一致的合到一塊，并輸出結果文件；如果有需要，采用Combiner進行最后的合并。
歸納來說主要分為5部分：用戶程序、Master、Mapper、Reducer、Combiner（上圖未給出）。
總的來說，架構不復雜。組件間通信用啥都可以，比如RPC、HTTP或者私有協議等。
2. 實現代碼介紹
該版本代碼實現了單機單進程版本，Mapper、Reducer和Combiner的實現用協程goroutine實現，通信采用channel。代碼寫的比較隨意，沒有解耦合。
為了方便起見，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. 總結
不足以及未實現之處：
接下來要是有空，我會實現分布式高可用的代碼，模塊間采用RPC通訊。
好了，以上就是這篇文章的全部內容了，希望本文的內容對大家的學習或者工作具有一定的參考學習價值，如果有疑問大家可以留言交流，謝謝大家對VEVB武林網的支持。
新聞熱點
疑難解答
