大數(shù)據(jù)職位

分布式數(shù)據(jù)處理框架

count the word frequency of a web page?

for循環(huán)，存在hashmap

缺點：只有一臺機器， 慢，內(nèi)存大小受限。

 

多臺機器，并行處理

合并的時候是瓶頸。

map把任務打散，reduce把任務合并

 

step1 input 輸入 

0: a b a c d d

1: a b c c d b

step2 split 輸入的拆分，給不同機器

m1 - 0:  a b a c d d

m2 - 1: a b c c d b

step3 map 機器分別執(zhí)行，不做aggregation

m1 - a,1 b,1 a,1 c,1 d,1 d,1

m2 - a,1 b,1 b,1 c,1 c1 d,1

step4 partition + sort

m1 - a,1 a,1 b,1 |  c,1 d,1 d,1

m2 - a,1 b,1 b,1 | c,1 d,1 d,1

step5 fetch + merge sort

m3 - a,1 a,1 b,1 | a,1 b,1 b,1

m4 - c,1 d,1 d,1 | c,1 d,1 d,1

m3 - a,[1,1,1]  b,[1,1,1]

m4 - c,[1,1,1]  d,[1,1,1]

step6 reduce 合起來

m3 - a,[3]  b,[3]

m4 - c,[3]  d,[3]

step7 output 輸出

a,[3]  b,[3] c,[3] d,[3]

 

step3不合并，不用hashmap

public static class Map {

　　public void map(String key, String value, OutputCollector<String, Integer> output) { // key 文章儲存地址，value文章內(nèi)容

　　　　// 切割文章中的單詞

　　　　StringTokenizer tokenizer = new StringTokenizer(value);

　　　　while (tokenizer.hasMoreTokens()) {

　　　　　　String outputkey = tokenizer.nextToken()；

　　　　　　output.collect(outputkey, 1);

　　　　}

　　}

　　public static class Reduce {

　　　　public void reduce(String key, Iterator<Integer> values, OutputCollector<String, Integer> output) { // key map輸出的key .. 

　　　　　　int sum = 0;

　　　　　　while (values.hasNext()) {

　　　　　　　　sum += values.next();

　　　　　　}

　　　　　　output.collect(key, sum);

　　　　}

　　}

}

 

partition and sort

master consistant hashing進行分組。硬盤上外排序

reduce把排好序的文件拿到對應的機器

 

map, reduce 多少機器。1000 + 1000

機器多，每臺處理的時間越少，總時間越快。啟動時間變長

reduce數(shù)目上限,key的數(shù)目

 

給定正倒排索引，建立倒排索引，給詞返回文章編號

 key 文章關鍵詞，value: 文章編號

reduce 去重操作，同一文章出現(xiàn)關鍵詞兩次的情況

// 同一文章下打散

public static class Map {

　　public void map (String key, Document value, OutputCollector<String, Integer> output) {

　　　　StringTokenizer tokenizer = new StringTokenizer(value.content);

　　　　while (tokenizer.hasMoreToken()) {

　　　　　　String word = tokenizer.nextToken();

　　　　　　output.collect(word, value.id);

　　　　}

　　}

}

// 同一單詞的合并

public static class Reduce {

　　public void reduce(String key, Iterator<Integer> values, OutputColllector<String, List<Integer>> output) {

　　　　List<Integer> results = new ArrayList<>();

　　　　int left = -1;

　　　　while (values.hasNext()) {

　　　　　　int now = values.next();

　　　　　　if (left != now) {

　　　　　　　　results.add(now);

　　　　　　}

　　　　　　left = now;

　　　　}

　　　　output.collect(key, results);

　　}

}

 

anagram:

map key: 每個單詞的root value: word

public static class Map{

　　public void map(String. key, String value, OutputCollector<String, String> output) {

　　　　StringTokenizer tokenizer = new StringTokenizer(value);

　　　　while (tokenizer.hasMoreTokens()) {

　　　　　　String word = tokenizer.nextToken();

　　　　　　char[] sc = word.toCharArray();

　　　　　　Arrays.sort(sc);

　　　　　　output.collect(new String(sc), word);

　　　　　　}

　　　}

}

reduce key：單詞  value：list

public static class Reduce {

　　public void reudce(String key, Iterator<String> values, OutputCollector<String, List<String>> output) {

　　　　List<String> results = new ArrrayList<>();

　　　　while (values.hasNext()) {

　　　　　　results.add(values.next());

　　　　}

　　　　output.collect(key, results);

　　}

}

 

top k frequency

class Pair {

　　String key;

　　int value;

　　Pair(String k, int v) {

　　　　key = k;

　　　　value = v;

　　}

　}

public void map(String _, Document value, OutputCollector<String, Integer> output) {

　　StringTokenizer tokenizer = new StringTokenizer(value.content);

　　　while (tokenizer.hasMoreTokens()) {
　　　　String word = tokenizer.nextToken();

　　　　output.collect(word, 1);

　　　}

}

public static class Reduce {

　　private PriorityQueue<Pair> Q;

　　private int k;

　　private Comparator<Pair> cmp = new Comparator<Pair>() {

　　　　public int compare(Pair a, Pair b) {

　　　　　　if (a.value != b.value) {

　　　　　　　　return a.value - b.value;

　　　　　　}

　　　　　　return b.key.compareTo(a.key);

　　　　}

　　};

　　public void setup(int k) {

　　　　Q = new PriorityQueue<Pair>(k, cmp);

　　　　this.key = k;

　　}

　　public void reduce(String key, Iterator<Integer> values) {

　　　　int sum = 0;　　

　　　　while (values.hasNext()) {

　　　　　　sum += values.next();

　　　　}

　　　　Pair cur = new Pair(key, sum);

　　　　if (Q.size() < k) {

　　　　　　Q.add(cur);

　　　　} else {

　　　　　　Pair peek = Q.peek();

　　　　　　if (cmp.compare(cur, peek) > 0) {

　　　　　　　　Q.poll();

　　　　　　　　Q.add(cur);

　　　　　　}

　　　　}

　　}

　　public void cleanup(OutputCollector<String, Integer> output) {

　　　　List<Pair> res = new ArrayList<>();

　　　　while (!Q.isEmpty()) {

　　　　　　res.add(Q.poll());

　　　　}

　　　　for (int i = res.size() - 1; i >= 0; i --) {

　　　　　　Pair cur = res.get(i);

　　　　　　output.collect(cur.key, cur.value);

　　　　}
　　}

}
 

design a MR system:

master 控制整個系統(tǒng)流程 - slave 完成真正的工作

1. 用戶指定多少map，多少reduce。啟動相應機器

2. master分配哪些slave作為map/ reduce。

3. master將input盡量等分給map, map讀取文件后執(zhí)行map工作

4. map工作后將結(jié)果寫到本地硬盤上

5. 傳輸整理將map結(jié)果傳給reduce

6. reduce工作，結(jié)束后將結(jié)果寫出

 

map結(jié)束了reduce

如果掛了一臺，重新分配一臺機器

reducer一個key特別多。加random后綴。類似shard key。 fb1, fb2, fb3分配到不同

input, output存放到GFS

local disk的mapper output data不需要保存GFS，丟了重做。中間數(shù)據(jù)不重要。

mapper和reducer之前有預處理，放在不同機器上

 

MapReduce whole process

1. start: user program start master and worker

2. assign task: master assign task to the map worker and reduce worker. assign map and reduce code

3. split: master split the input data

4. map read: each map worker read the split input data

5. map: each map worker do the map job on their machine

6. map output: each map worker output the file in the local disk of its worker

6. reduce fetch: each reduce worker fetch the data from the map worker

7. reduce: each reducer worker do the reduce job on their machine

8. reduce output: reduce worker outpt the final output data