Skip to content

lab01.md

Latest commit

 

History

History
157 lines (112 loc) · 4.76 KB

lab01.md

File metadata and controls

157 lines (112 loc) · 4.76 KB
Raw

6.824 Lab 1: MapReduce

目录：

细分目录：

所有的 test 都过了

总体而言没什么难度。

大体思路

  • Master 进程负责管理各个任务的状态、分配任务
  • Worker 执行后，根据哈希保存到相应路径，再向 Master 发送消息
    • “我做完了”，“你可以更新下这个任务的状态了”
    • “我想要一个新任务”，“你可以在 response 里给我回复一个新任务”

注意 Master 是一个服务， Worker 主动向 Master 发生请求。

用 socket + go rpc 协议交互。

与论文的区别

大概读了下论文，最大的区别在于：

  • 论文中 Master 还记录了需要 Reduce 文件 list ，但是我是基于文件命名规则在 Worker 中 glob 的

一些其他收获

go: compile to so file AND load so file

go build $RACE -buildmode=plugin wc.go

# load
../mrsequential ../../mrapps/wc.so

In mrsequential codes:

mapf, reducef := loadPlugin(os.Args[1])

防止文件操作冲突，先写到 temp/xxx 文件中，写完再拷贝

这点论文 4.5 和作业提示中都提到了。

似乎 linux 中拷贝这个过程是具备原子性的。

Map+Reduce 总体给我的感觉就是分布式 groupby ？

或者说可以套用 groupby 框架？

比如论文中的 2.3 More Examples 章节，提到的事例，都是 groupby 。

map    (k1, v1)       -> list(k2, v2)
reduce (k2, list(v2)) -> list(v2)

比如 wc.go

func Map(filename string, contents string) []mr.KeyValue {
	// function to detect word separators.
	ff := func(r rune) bool { return !unicode.IsLetter(r) }

	// split contents into an array of words.
	words := strings.FieldsFunc(contents, ff)

	kva := []mr.KeyValue{}
	for _, w := range words {
		kv := mr.KeyValue{w, "1"}
		kva = append(kva, kv)
	}
	return kva
}

//
// 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 Reduce(key string, values []string) string {
	// return the number of occurrences of this word.
	return strconv.Itoa(len(values))
}

值得注意的是，把相同 key 对应的 values 这一行为“聚合”起来这一行为，并不由 reduce 来处理，而是在 reduce 前， sort by key ，然后得到各个 key 的 list of value 。

我是否能将其转换为 groupby 呢？我尝试下：

from typing import List, Tuple
import pandas as pd
from itertools import groupby, chain

# 这里之所以留着 filename ，大概是为了保留全量 input 信息，比如 indexer 时构建 values 时会用到
def Map(filename: str, contents: str) -> List[Tuple[str, str]]:
    lis = contents.split(' ')
    return [(x, '1') for x in lis]

def Reduce(key: str, values: List[str]) -> str:
    return sum([int(x) for x in values])

meta_data = [
    {'filename': 'whale', 'contents': 'a lot of words'},
    {'filename': 'horse', 'contents': 'a little speak'},
]

# map
intermediate = map(lambda x: Map(x['filename'], x['contents']), meta_data)
# merge data for keys
# 在实操中，并不会把所有 keys 的数据都 merge ，然后排序
# 而是也会根据 hash 把 keys 分组，然后 load ，然后排序， reduce
intermediate = chain.from_iterable(intermediate)
intermediate = list(intermediate)
for key, key_values in groupby(  # 注意这里需要 sorted,
        # https://docs.python.org/3/library/itertools.html
        # groupby 需要 sort 才能全部合并 key
        sorted(intermediate, key=lambda x: x[0]),
        lambda x: x[0]):
    values = [x[1] for x in key_values]
    print(key, Reduce(key, values))

Scala 中 groupMap 与 groupMapReduce

还有一篇 scala 文章：