map reduce案例超详细讲解

实验一：对输入文件统计单词频率

A good beginning is half the battle,
Where there is a will there is a way.
There is no royal road to learning.
1
2
3

map输出，分割单词增加尾部1，方便统计

import sys

for line in sys.stdin:
	ss = line.strip().split(' ')
	for word in ss:
		print('\t'.join([word.strip(),'1']))

1
2
3
4
5
6
7

reduce统计

import sys

cur_word = None
sum = 0

for line in sys.stdin:
	word,cnt = line.strip().split('\t')

	if cur_word == None:
		cur_word = word

	if cur_word != word:
		print('\t'.join([cur_word,str(sum)]))
		cur_word = word
		sum = 0

	sum += int(cnt)
	
print('\t'.join([cur_word,str(sum)]))
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19

调用管道符测试

cat data.txt | python map.py | sort -k 1 | python reduce.py
1

上面是一个简单的例子
加大一点点难度
使用MapReduce实现如下功能(要mapreduce思维实现，考虑多节点协同处理，非以往的Python单节点处理数据思维)

现有数据 emp.csv 、 dept.txt
emp.csv ----------------------------- ------------------------ dept.csv
部门编号 姓名 年龄 -------------------- 部门编号 部门名称
20 zhaoyi 30 ----------------------------------- 20 技术部
30 liuer 25 ----------------------------------- 10 人事部
10 zhangsan 31 ----------------------------------- 30 财务部
20 lisi 40
30 wangwu 30
20 sunliu 22

最终要输出的如下

部门名称 人员列表
人事部 zhangshan:31
技术部 zhaoyi-30 ,lisi-40, sunliu–22
财务部 liuer-25,wangwu—30

这个是找的答案的

# Map：
import sys 
import io 
input_str = io.TextIOWrapper(sys.stdin.detach(),encoding='utf-8 ')
for line in input_str: 
     datas = line.strip().split(',')
     if(len(datas)==3):
           dept = datas[0]
           name = datas[1]+'-'+datas[2]
           print(dept+'\t'+name)
     if(len(datas)==2):
            dept = datas[0]
            name = datas[1]
            print(dept+'\t'+'*'+name)






# Reduce：
import sys 
import io 
input_str = io.TextIOWrapper(sys.stdin.detach(),encoding='utf-8 ')
cur_dept = None 
emp_list='' 
dept_name=''
for line in input_str: 
        dept, name = line.strip().split('\t')
        if cur_dept == None:
            cur_dept = dept
        if cur_dept != dept:
            print('\t'.join([dept_name, emp_list]))
            cur_dept = dept 
            emp_list = ''
        if "*" in name:
            dept_name = name[1:]
        else:
            emp_list += name + ','
            print('\t'.join([dept_name, emp_list]))

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41

这个过程嘛……上面那个哟有点长我丢

按照这个分步骤一下
map的输入

cat demo.csv | python map2.py 
1

先用一个csv把他们装起来
首先第一步我们在虚拟机上创建了文件：demo.csv 这个文件是题目两个文件合起来的，也可以用管道符合起来这里为了方便就没有用

文件的内容

20,zhaoyi,30
30,liuer,25
10,zhangsan,31
20,lisi,40
30,wangwu,30
20,sunliu,22
20,技术部
10,人事部
30,财务部

1
2
3
4
5
6
7
8
9
10

我们观察发现，第一行，到第六行，它有部门编号，名字，年龄
第七行到第九行是编号和部门名字
所以他们的区别在于长度不一样

#我们首先把map接受的东西区分一下
#1、第一他是有要分开的，一个是人的名字，一个是部门的名字
#2.增加一些东西给他去分开

import sys
for line in sys.stdin:
    print(line)#我们可以看看数据是怎样的

1
2
3
4
5
6
7
8

    #line我们需要进行分割
    line = line.strip().split(",")#按照','分割
    #第一步取出部门编号 和人名
    #区别
    print(len(line))
    #line的长度，带人名的是3 部门的是2

1
2
3
4
5
6
7

完整map代码

#我们首先把map接受的东西区分一下
#1、第一他是有要分开的，一个是人的名字，一个是部门的名字
#2.增加一些东西给他去分开

import sys
for line in sys.stdin:
    #line我们需要进行分割
    line = line.strip().split(",")#按照','分割
    #第一步取出部门编号 和人名
    #区别
    #line的长度，带人名的是3 部门的是2
    #分类
    if len(line)==3:#带人名的是3
        #按照题目要求
        #输出--ID--------------名字-------------年龄
        print(line[0] + "\t" + line[1] + "-" + line[2])
    else:#部门的是2
        #这里用*和他们分开
        #输出--ID--------------*--部门名字
        print(line[0]+"\t"+'*'+line[1])

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21

map的输出

对输出排序后交给reduce2处理

cat demo.csv | python map2.py |sort -k 1 |python reduce2.py
1

首先我们创建容器来装我们要的数据

id_name = None#这个用来装 ID
bm_name=" "#这个用来装     部门的 名字
yg_name=" "#这个用来装     员工的 名字
1
2
3

接着我们对k3,v3读取

for line in sys.stdin:
    #老规矩我们查看一下这个line，发现管道符命令已经对他排好序了，而且部门是放在最后一位的
    #k3,            v3
    # id(部门id)  massage(部门名字人名年龄那些)
    # 10         zhangsan - 31
    # 10           * 人事部
    id,massage = line.strip().split('\t')  #k3,v3创建
1
2
3
4
5
6
7

#第一步
#记录ID 确定输出的时候（什么时候完成了一个部门的总结就输出）

记录一个id

    if id_name==None:#也就是这个id_name还没有任何数据的时候
        id_name = id#那就把这个ID记录到这个id_name里面
1
2

确定输出的时候

f id_name != id:#这个是判断新的ID他是一个个读取的
        #例如读取到下面上一个的ID和下一个不一样的时候
        #10      *人事部
        #20      lisi-40
        #也就是编号10的部门以及读取完了这个时候对吧，所以他的10就可以输出了
        #输出       部门名字           员工信息
        print(str(bm_name)+"\t"+str(yg_name))
1
2
3
4
5
6
7

记录下一个id

        id_name = id  # 那就把这个不一样的ID记录到这个id_name里面
        #然后别忘记把员工的信息清空，因为你总不可能记录这个部门还保存着上一个部门的员工信息，这样到时候输出就不好输出了
        yg_name = " "
1
2
3

#第二步骤
#通过ID 添加信息
#部门 + 人的名字-年龄

#第二步
    if "*" in massage:
        bm_name = massage[1:]
    else:
        yg_name += massage+';'#这里我就解释一下为什么这两个加法不一样，首先我们输出的是 部门1 员工1；员工2； 是这样的部门是直接替换原来的但是员工是累加的
1
2
3
4
5

完整代码

import sys
id_name = None#这个用来装 ID
bm_name=" "#这个用来装     部门的 名字
yg_name=" "#这个用来装     员工的 名字
for line in sys.stdin:
    id,massage = line.strip().split('\t')
    if id_name==None:
        id_name = id
    if id_name != id:
        print(str(bm_name)+"\t"+str(yg_name))
        id_name = id 
        yg_name = " "
    if "*" in massage:
        bm_name = massage[1:]
    else:
        yg_name += massage+';'
print(str(bm_name) + "\t" + str(yg_name))
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17

这里最后这个输出为什么在外面呢，首先第一点是，当记录最后一个部门的时候
开始记录这个部门
20 --------------------------*技术部--------------------------
也就是读到部门编号不一样的时候，20-30的时候，他会做判断 if id_name != id:，然后输出这个部门的所有东西
30 -------------------------- liuer-25-----------从这里开始新的记录
30 -------------------------- wangwu-3030 *财务部--------------记录完成
但是下面已没有不同的id支持他输出了
于是利用for循环的一个特性输出，（遍历完成后输出最后一个）
这个的理解需要我们来进行一个小实验

list=[1,2,3]
for i in list:
      print('猜猜我在外面会输出一个怎么样的i')
 print(i)
输出 3也就是最后一组
1
2
3
4
5

总结reduce