python 实现MR

1. 看下本地的测试数据：

[root@hadoop Desktop]# cat tour.txt
air:23;hotel:34;nation:CHINA
air:35;hotel:46;nation:USA
air:36;hotel:47;nation:USA
air:26;hotel:37;nation:CHINA
air:33;hotel:44;nation:USA
air:34;hotel:45;nation:USA
air:25;hotel:36;nation:CHINA
air:24;hotel:35;nation:CHINA
[root@hadoop Desktop]# cat mapper.txt
cat: mapper.txt: No such file or directory

2. 设计mapper

[root@hadoop Desktop]# cat mapper.py
#!/usr/bin/python

import sys

for line in sys.stdin:
    airs=line.split(";")
    air=airs[0].split(":")
    hotel=airs[1].split(":")
    print hotel[0]+" "+hotel[1]
    print air[0]+" "+air[1]

3. 设计reducer

[root@hadoop Desktop]# cat reducer.py
#!/usr/bin/python

import sys

allsum={}
product=""

for line in sys.stdin:
    products=line.split()
    product=products[0]
    allsum[product]=allsum.get(product,0)+int(products[1])

for key,value in allsum.items():
print key+" "+str(value)

4. 赋予mapper.py 和reducer.py 执行权限

5. 测试mapper

[root@hadoop Desktop]# cat tour.txt | ./mapper.py
hotel 34
air 23
hotel 46
air 35
hotel 47
air 36
hotel 37
air 26
hotel 44
air 33
hotel 45
air 34
hotel 36
air 25
hotel 35
air 24

5.测试reducer

[root@hadoop Desktop]# cat tour.txt | ./mapper.py | ./reducer.py
hotel 324
air 236

6.将测试数据上传到HDFS

[root@hadoop Desktop]# hadoop fs -ls /usr/egencia/tour/tour.txt
/usr/hadoop/hadoop-1.2.1/libexec/../conf/hadoop-env.sh: line 59: export: `mapred.tasktracker.reduce.tasks.maximum=4': not a valid identifier
Warning: $HADOOP_HOME is deprecated.

Found 1 items
-rw-r--r-- 1 root supergroup 224 2013-08-27 05:22 /usr/egencia/tour/tour.txt

7. 在hadoop 执行MR

[root@hadoop Desktop]# hadoop jar /usr/hadoop/hadoop-1.2.1/contrib/streaming/hadoop-streaming-1.2.1.jar -mapper /root/Desktop/mapper.py -reducer /root/Desktop/reducer.py -input /usr/egencia/tour -output /usr/egencia/tour/out

注意：执行的时候在指定mapper和reducer的时候不能像在本地测试一样采用./mapper.py 和./reducer.py ,这也是容易理解的，因为在将这两个文件拷贝到其他的datanode上去的时候，默认的执行目录不一定就是我在本地测试的desktop

8. HDFS 上查看结果：

[root@hadoop Desktop]# hadoop fs -cat /usr/egencia/tour/out/part-00000
/usr/hadoop/hadoop-1.2.1/libexec/../conf/hadoop-env.sh: line 59: export: `mapred.tasktracker.reduce.tasks.maximum=4': not a valid identifier
Warning: $HADOOP_HOME is deprecated.

hotel 324
air 236

可以看出结果和本地测试的是一样的，但是在python版本中怎么实现partitioner ，combiner，等还需考量，另外从上面的reducer'的函数也可以看出，其实根本就不会像java一样输入输出都是key、value而是自己可以随便定义，当然也不会自动将同一个key组合的一个reducer中作为一个迭代，所以在上面的reducer中药自己控制迭代，这些问题待解

python 实现MR相关推荐

python项目开发：ftp server开发
程序要求: 1.用户加密认证 (对用户名密码进行MD5验证)2.允许同时多用户登陆 (使用socket server方法,为每个用户都创建一个信息文件)3.每个用户有自己的家目录,且只能访问自己的家目 ...
hadoop streaming编程小demo(python版)
大数据团队搞数据质量评测.自动化质检和监控平台是用django,MR也是通过python实现的.(后来发现有orc压缩问题,python不知道怎么解决,正在改成java版本) 这里展示一个python ...
streaming mr
在以前的一篇关于python版MR实例的文章中,只是简单走了一下流程,这次主要是解决上次遗留的key,partition,sort的问题.以进一步理解hadoop streaming,也尝试mappe ...
吐血整理出来的大数据知识点，你掌握多少？
前言 hi!我是啊晨 ,一个大数据分享者,好久没发博客了,最近忙于找工作,然后顺带整理了一些资料,相信也有一部分好兄弟,也在苦于找工作,或者正在学习大数据的知识,不太会总结的.我只能这样说,你来对了. ...
Apache DolphinScheduler 大数据工作流调度系统
[本文正在参与 "拥抱开源 - Apache DolphinScheduler 有奖征稿活动],活动地址 Apache DolphinScheduler 大数据工作流调度系统一.背景二. ...
Apache DolphinScheduler 2.X保姆级源码解析，中国移动工程师揭秘服务调度启动全流程...
2022年1月,科学技术部高新技术司副司长梅建平在"第六届中国新金融高峰论坛"上表示,当前数据量已经大大超过了处理能力的上限,若信息技术仍然是渐进式发展,则数据处理能力的提升将远远 ...
4 亿用户，7W+ 作业调度难题，Bigo 基于 Apache DolphinScheduler 巧化解
点击上方蓝字关注我们 ✎ 编者按成立于 2014 年的 Bigo,成立以来就聚焦于在全球范围内提供音视频服务.面对 4 亿多用户,Bigo 大数据团队打造的计算平台基于 Apache Dolp ...
Python OpenCV开发MR智能人脸识别打卡系统(一、需求分析与系统设计)
需要源码请点赞关注收藏后评论区留言私信~~~ 整体系统讲解如下链接 Python OpenCV开发MR智能人脸识别打卡系统(二.文件系统.数据实体模块设计) Python OpenCV开发MR智能人脸 ...
Python OpenCV开发MR智能人脸识别打卡系统（四、服务模块设计）
需要源码请点赞关注收藏后评论区留言私信~~~ 整体系统讲解如下 Python OpenCV开发MR智能人脸识别打卡系统(一.需求分析与系统设计) Python OpenCV开发MR智能人脸识别打卡系统 ...

python 实现MR

python 实现MR相关推荐

最新文章

热门文章