rowcount-gist.go

package main

import (
	"fmt"
	"log"
	"os"
	"runtime"
	"strconv"
	"strings"
	"sync"

	"github.com/cnnrznn/gomr"
)

type Job struct{}

type Entry struct {
	key int
	val string
}

func (j *Job) Map(in <-chan interface{}, out chan<- interface{}) {
	defer close(out)

	for item := range in {
		phone, err := strconv.Atoi(strings.Split(item.(string), ",")[0])
		if err != nil {
			log.Fatal(err)
		}
		out <- Entry{key: phone, val: item.(string)}
	}
}

func (j *Job) Partition(in <-chan interface{}, outs []chan interface{}, wg *sync.WaitGroup) {
	defer wg.Done()
	for item := range in {
		key := item.(Entry).key
		outs[key%len(outs)] <- item
	}
}

func (j *Job) Reduce(in <-chan interface{}, out chan<- interface{}, wg *sync.WaitGroup) {
	defer wg.Done()
	counts := make(map[int][]string)

	for item := range in {
		entry := item.(Entry)
		counts[entry.key] = append(counts[entry.key], entry.val)
	}

	for _, v := range counts {
		out <- v
	}
}

func main() {
	job := &Job{}
	p := runtime.NumCPU()

	ins, out := gomr.RunLocal(2*p, p, job)
	gomr.TextFileParallel(os.Args[1], ins)

	for item := range out {
		fmt.Println(item)
	}
}