wc.go
package main
import (
"fmt"
"mapreduce"
"os"
"strings"
"unicode"
"strconv"
)
//
// The map function is called once for each file of input. The first
// argument is the name of the input file, and the second is the
// file's complete contents. You should ignore the input file name,
// and look only at the contents argument. The return value is a slice
// of key/value pairs.
//
func mapF(filename string, contents string) []mapreduce.KeyValue {
// Your code here (Part II).
f := func(c rune) bool {
return !unicode.IsLetter(c)
}
rst := make([]mapreduce.KeyValue, 0)
keys := strings.FieldsFunc(contents, f)
for _, key := range keys {
kv := mapreduce.KeyValue{ Key: key, Value:"1"}
rst = append(rst, kv)
}
return rst
}
//
// The reduce function is called once for each key generated by the
// map tasks, with a list of all the values created for that key by
// any map task.
//
func reduceF(key string, values []string) string {
// Your code here (Part II).
cnt := 0
for _ , val := range values {
i , err := strconv.Atoi(val);
if err != nil {
panic(err)
}
cnt += i
}
rst := strconv.Itoa(cnt)
return rst
}
// Can be run in 3 ways:
// 1) Sequential (e.g., go run wc.go master sequential x1.txt .. xN.txt)
// 2) Master (e.g., go run wc.go master localhost:7777 x1.txt .. xN.txt)
// 3) Worker (e.g., go run wc.go worker localhost:7777 localhost:7778 &)
func main() {
if len(os.Args) < 4 {
fmt.Printf("%s: see usage comments in file\n", os.Args[0])
} else if os.Args[1] == "master" {
var mr *mapreduce.Master
if os.Args[2] == "sequential" {
mr = mapreduce.Sequential("wcseq", os.Args[3:], 3, mapF, reduceF)
} else {
mr = mapreduce.Distributed("wcseq", os.Args[3:], 3, os.Args[2])
}
mr.Wait()
} else {
mapreduce.RunWorker(os.Args[2], os.Args[3], mapF, reduceF, 100, nil)
}
}