MapReduce过程

MapReduce是采用一种分而治之的思想设计出来的分布式计算框架，它由两个阶段组成：map阶段和reduce阶段。​

在map阶段中：

• 首先读取HDFS中的文件，每个文件都以一个个block形式存在，block中的数据会被解析成多个kv对，然后调用map task的map方法；
• map方法对接收到的kv对进行分片处理，转换输出成新的kv对；
• 然后对kv对进行分区和排序压缩，中间会涉及到map的shuffle过程，最后存入本地磁盘，供后续reduce task作为输入参数使用。

在reduce阶段中：

• reduce方法将多个map task的输出，按照不同的分区传输到不同的reduce 上进行合并排序，这中间也涉及到reduce的shuffle过程(shuffle就像洗扑克牌一样)
• 然后reduce方法对输入的键值对汇总计算，输出计算结果
• 最后把reduce的输出保存在HDFS文件中。

Linux shell统计文章中出现频次最高的单词：

]# cat 文件名| grep -Eo --color '\w+' | sort | uniq -c | sort -k1 -nr | head -1

解释：-e 使用正则 -o 只打出匹配到的字符 \w+ 匹配字母

uniq 去重 ，uniq -c 打印频次,会显示两列 第一列 频次 ，第二列数据

sort -k1 代表第一列 -n按数字排序 ，默认由小到大 -nr由大到小排序

#!/usr/bin/env python
import sys
for line in sys.stdin:   line = line.strip()   words = line.split()   for word in words:     print "%s\t%s" % (word, 1)
#!/usr/bin/env python
from operator import itemgetter
import syscurrent_word = Nonecurrent_count = 0
word = None
for line in sys.stdin:  line = line.strip()  word, count = line.split('\t', 1) try:    count = int(count)  except ValueError:    continue  if current_word == word:   current_count += count  else:    if current_word:      print "%s\t%s" % (current_word, current_count)    current_count = count    current_word = word if word == current_word:print "%s\t%s" % (current_word, current_count)

map.pyimport sys
for line in sys.stdin: ss = line.strip().split('\t')
if len(ss) != 2:
continue
file_name, file_content = ss
word_list = file_content.strip().split(' ')
word_set = set(word_list)
for word in word_set:
print '\t'.join([word, '1'])red.py[root@master tfidf_test]# cat red.py
import sysimport math
current_word = Nonesum = 0
docs_cnt = 508
for line in sys.stdin:  ss = line.strip().split('\t')  if len(ss) != 2:    continue  word, val = ss  if current_word == None:    current_word = word  if current_word != word:   idf = math.log(float(docs_cnt) / (float(sum) + 1.0))   print '\t'.join([current_word, str(idf)])   current_word = word    sum = 0   sum += int(val)idf = math.log(float(docs_cnt) / (float(sum) + 1.0))print '\t'.join([current_word, str(idf)])输出当前单词+idf分