前言
MapReduce作為hadoop的編程框架,是工程師最常接觸的部分,也是除去了網絡環(huán)境和集群配 置之外對整個Job執(zhí)行效率影響很大的部分,所以很有必要深入了解整個過程。元旦放假的第一天，在家沒事干，用golang實現了一下mapreduce的單進程版本，github地址。處理對大文件統(tǒng)計最高頻的10個單詞，因為功能比較簡單，所以設計沒有解耦合。
本文先對mapreduce大體概念進行介紹，然后結合代碼介紹一下，如果接下來幾天有空，我會實現一下分布式高可用的mapreduce版本。下面話不多說了，來一起看看詳細的介紹吧。
1. Mapreduce大體架構
上圖是論文中mapreduce的大體架構。總的來說Mapreduce的思想就是分治思想：對數據進行分片，然后用mapper進行處理，以key-value形式輸出中間文件；然后用reducer進行對mapper輸出的中間文件進行合并：將key一致的合到一塊，并輸出結果文件；如果有需要，采用Combiner進行最后的合并。
歸納來說主要分為5部分：用戶程序、Master、Mapper、Reducer、Combiner（上圖未給出）。
- 用戶程序。用戶程序主要對輸入數據進行分割，制定Mapper、Reducer、Combiner的代碼。
- Master：中控系統(tǒng)。控制分發(fā)Mapper、Reduer的個數，比如生成m個進程處理Mapper，n個進程處理Reducer。其實對Master來說，Mapper和Reduer都屬于worker，只不過跑的程序不一樣，Mapper跑用戶輸入的map代碼，Reduer跑用戶輸入的reduce代碼。Master還作為管道負責中間路徑傳遞，比如將Mapper生成的中間文件傳遞給Reduer，將Reduer生成的結果文件返回，或者傳遞給Combiner（如果有需要的話）。由于Master是單點，性能瓶頸，所以可以做集群：主備模式或者分布式模式?？梢杂脄ookeeper進行選主，用一些消息中間件進行數據同步。Master還可以進行一些策略處理：比如某個Worker執(zhí)行時間特別長，很有可能卡住了，對分配給該Worker的數據重新分配給別的Worker執(zhí)行，當然需要對多份數據返回去重處理。
- Mapper：負責將輸入數據切成key-value格式。Mapper處理完后，將中間文件的路徑告知Master，Master獲悉后傳遞給Reduer進行后續(xù)處理。如果Mapper未處理完，或者已經處理完但是Reduer未讀完其中間輸出文件，分配給該Mapper的輸入將重新被別的Mapper執(zhí)行。
- Reducer: 接受Master發(fā)送的Mapper輸出文件的消息，RPC讀取文件并處理，并輸出結果文件。n個Reduer將產生n個輸出文件。
- Combiner: 做最后的歸并處理，通常不需要。
總的來說，架構不復雜。組件間通信用啥都可以，比如RPC、HTTP或者私有協議等。
2. 實現代碼介紹
該版本代碼實現了單機單進程版本，Mapper、Reducer和Combiner的實現用協程goroutine實現，通信采用channel。代碼寫的比較隨意，沒有解耦合。
- 功能：統(tǒng)計給定文件中出現的最高頻的10個單詞
- 輸入：大文件
- 輸出：最高頻的10個單詞
- 實現：5個Mapper協程、2個Reducer、1個Combiner。
為了方便起見，Combiner對最高頻的10個單詞進行堆排序處理，按規(guī)范來說應該放在用戶程序處理。
文件目錄如下，其中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.go
6 directories, 8 files
2.1 caller
用戶程序，讀入文件并按固定行數進行劃分；然后調用master.Handle進行處理。
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，處理消息中轉，輸出最后結果。
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傳遞過來的中間文件并歸并。
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傳遞過來的Reducer結果文件并歸并成一個，然后堆排序輸出最高頻的10個詞語。
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
}
3. 總結
不足以及未實現之處：
- 各模塊間耦合性高
- master單點故障未擴展
- 未采用多進程實現，進程間采用RPC通信
- 未實現單個Workder時間過長，另起Worker執(zhí)行任務的代碼。
接下來要是有空，我會實現分布式高可用的代碼，模塊間采用RPC通訊。
好了，以上就是這篇文章的全部內容了，希望本文的內容對大家的學習或者工作具有一定的參考學習價值，如果有疑問大家可以留言交流，謝謝大家對腳本之家的支持。
您可能感興趣的文章:- golang 輸出重定向:fmt Log,子進程Log,第三方庫logrus的詳解
- Golang信號處理及如何實現進程的優(yōu)雅退出詳解
- golang守護進程用法示例
- golang 后臺進程的啟動和停止操作
