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执行任务的代码。接下来要是有空，我会实现分布式高可用的代码，模块间采用RPC通讯。
好了，以上就是这篇文章的全部内容了，希望本文的内容对大家的学习或者工作具有一定的参考学习价值，如果有疑问大家可以留言交流，谢谢大家对易采站长站的支持。
