MapReduce程序实例（python）

男娘i 2022-02-25 13:58 201阅读 0赞

### 问题背景 ###

**现在有两份数据，file1是校园新闻版块，每一条新闻点击记录；file2是校园新闻版块使用活跃度高的学生记录。用mr统计出某一天的点击记录里，使用ios/android手机的活跃学生的总的点击次数**  
原始数据格式如下：  
**file1.log，校园新闻点击记录**，记录了每一次学生点击校园新闻的行为  
格式：学号\\t姓名\\t手机端\\t新闻id\\新闻关键词

> 20170001 xiaoming android 331 学费  
> 20170002 xiaohong ios 332 食堂  
> 20170003 xiaohua android 333 考研  
> 20170001 xiaoming android 222 评优  
> 20170001 xiaoming android 225 学费  
> 20170003 xiaohua android 111 期末考试  
> 20170002 xiaohong ios 117 空调安装

**file2.log，校园新闻活跃学生记录**，记录了对校园新闻版块点击活跃度高的学生信息  
格式：格式：学号\\t姓名\\t年龄\\t手机端

> 20170001 xiaoming 20 android  
> 20170002 xiaohong 19 ios  
> 20170001 xiaoqiang 20 android

### 问题分析 ###

1、key的选择：选取学号+手机端共同作为key。学号是用于对是否活跃用户的判断，手机端是用于统计时的分类。**注意**，这里如果仅仅将手机端作为key有一个坏处，就是key值将只有两种，这样不利于将海量的数据分为多个小部分处理，因此选择学号+手机端共同作为key。  
2、关键点：在对每一行数据进行处理时，利用两份数据，做好是否活跃用户的标记；在对同一个key进行处理时，根据前一步处理的结果，判断是否进行累加。

### 代码实现 ###

**1、mapper.py**  
解析原始数据，将其处理为一系列的键值对（key-value pair），这里选取"学号\_手机端"作为key。

> \#!/usr/bin/env python  
> \# -\*- coding:utf-8 -\*-
> 
> import sys
> 
> def main():  
>     """  
>     mapper  
>     """  
>     for line in sys.stdin:  
>         data = line.strip().split(" ")  
>         if len(data) == 5:  
>             student\_id = data\[0\]  
>             os\_name = data\[2\]  
>             key = student\_id + '\_' + os\_name  
>             print"\\t".join(\[key,"file1"\])  
>         elif len(data) == 4:  
>             student\_id = data\[0\]  
>             os\_name = data\[-1\]  
>             key = student\_id + '\_' + os\_name  
>             print"\\t".join(\[key,"file2"\])  
> \#if \_\_name\_\_ == "\_\_main\_\_":  
> \#    main()  
> main()

**2、reducer.py**  
由MapReduce框架的机制，同一个key的所有数据都会被传给同一个reducer，同一个reducer可以接受多个键值对。  
reducer将相同键的一组值，进行处理产生一组规模更小的值（如累加得到一个值）。

> \#!/usr/bin/env python  
> \# -\*- coding:utf-8 -\*-  
> import sys  
> import json
> 
> dic\_result = \{'android': 0, 'ios': 0\}  
> pre\_key = ""
> 
>   
> def post\_deal(pre\_key):  
>     \# print pre\_key  
>     \# 相同key的数据的处理  
>     global dic  
>     os\_name = pre\_key.strip().split("\_")\[1\]  
>     \# print dic  
>     \# if 'click' in dic and dic\['click'\] == 'yes' :\#and 'active' in dict and dic\['active'\] == 'yes':  
>     if 'click' in dic and dic\['click'\] == 'yes' and 'active' in dic and dic\['active'\] == 'yes':
> 
>         \# if os\_name not in dic\_result:  
>         \#     dic\_result\[os\_name\] =0  
>         dic\_result\[os\_name\] += 1
> 
>   
> def deal(data):  
>     \# 对每一行数据的处理  
>     if data\[1\].strip() == "file1":  
>         dic\['click'\] = 'yes'  
>     if data\[1\].strip() == "file2":  
>         dic\['active'\] = 'yes'
> 
>   
> def pre\_deal():  
>     \# 预处理，如将用于存储同一个key的一组value中间结果的字典置零  
>     global dic  
>     dic = dict()
> 
>   
> def main():  
>     \# reducer  
>     \# data\_list = \['20170001\_android    file1','20170002\_ios    file1','20170001\_android    file1','20170001\_android    file1', '20170001\_android   file2','20170001\_android    file2',  
>     \#              '20170002\_ios    file1','20170002\_ios    file2','20170003\_android    file1', '20170003\_android   file1'\]  
>     \# for line in data\_list:  
>     for line in sys.stdin:  
>         data = line.strip().split("\\t")  
>         \# print data
> 
>         key = data\[0\]  
>         global pre\_key  
>         \# 当同一个key的数据全部deal完，到下一个key时  
>         if key != pre\_key:  
>             if pre\_key != "":  
>                 \# 处理同一个key所有数据  
>                 post\_deal(pre\_key)  
>             \# 处理完成后，预处理准备下一轮处理  
>             pre\_deal()  
>             \# 将pre\_key更新为下一个要处理的key  
>             pre\_key = key
> 
>         \# 将同一个key的一组值按需要进行处理/整合  
>         deal(data)
> 
>         \# 处理最后一个key的数据  
>         if pre\_key != "":  
>             post\_deal(pre\_key)  
>         \# 打印结果  
>     print json.dumps(dic\_result)
> 
>   
> \#if \_\_name\_\_ == "\_\_main\_\_":  
> \#    main()  
> main()

**3、线下测试 cat file|map|sort|reduce**  
STEP1：测试map  
`cat file1.log file2.log | python mapper.py > map_out`

`或者 cat *log |python mapper.py > map_out`  
测试中，map\_out中的数据将如下所示：

> $ more map\_out   
> 20170001\_android    file1  
> 20170002\_ios    file1  
> 20170003\_android    file1  
> 20170001\_android    file1  
> 20170001\_android    file1  
> 20170003\_android    file1  
> 20170002\_ios    file1  
> 20170001\_android    file2  
> 20170002\_ios    file2  
> 20170001\_android    file2

**测试：**

> $ cat \*log | python mapper.py |sort -k1,1  
> 20170001\_android    file1  
> 20170001\_android    file1  
> 20170001\_android    file1  
> 20170001\_android    file2  
> 20170001\_android    file2  
> 20170002\_ios    file1  
> 20170002\_ios    file1  
> 20170002\_ios    file2  
> 20170003\_android    file1  
> 20170003\_android    file1

STEP2：对map\_out按照key（key默认放在第一列）排序，使相同key的数据在一起，然后测试reducer  
`cat map_out |sort -k1,1|python reducer.py > red_out`  
reducer的输出是json，形如：\{'ios':1123, 'android':3333\}

> $ cat map\_out | sort -k1,1|python reducer.py   
> \{"android": 3, "ios": 2\}

**4、hadoop线上运行 run.sh**

> source ~/.bashrc
> 
> cwd=$(cd $(dirname $0); pwd)
> 
> input="/home/hdfs/project/test\_hadoop\_streaming/input/\*log"  
> output="/home/hdfs/project/test\_hadoop\_streaming/output/result"
> 
> \# 测试Hadoop streaming程序  
> jobname="yaoyingzhe\_hadoop\_streaming\_$thedate"  
> hadoop fs -rmr $output  
> hadoop streaming \\  
>     -D mapred.job.name="$jobname" \\  
>     -D mapred.output.compress=true \\  
>     -D mapred.output.compression.codec=org.apache.hadoop.io.compress.GzipCodec \\  
>     -input $input \\  
>     -output $output \\  
>     -mapper "mapper.py" \\  
>     -reducer "reducer.py"  \\  
>     -file $cwd/mapper.py\\  
>     -file $cwd/reducer.py

部分参数的解释：  
mapred.map.tasks：map数目  
mapred.reduce.tasks：reduce数目  
mapred.job.map.capacity：map并发数目  
mapred.job.reduce.capacity：educe并发数目  
stream.num.map.output.key.fields：设置map程序分隔符的位置，该位置之前的部分作为key，之后的部分作为value

参考：[https://www.cnblogs.com/surimj/p/9863841.html][https_www.cnblogs.com_surimj_p_9863841.html]

[https_www.cnblogs.com_surimj_p_9863841.html]: https://www.cnblogs.com/surimj/p/9863841.html