文档章节

hadoop平台wordcount程序的python实现

潘少online
 潘少online
发布于 2015/06/06 18:26
字数 1334
阅读 99
收藏 2

例子的目的是统计输入文件的单词的词频。

  • 输入:文本文件

  • 输出:文本(每行包括单词和单词的词频,两者之间用'\t'隔开)

 

1. Python MapReduce 代码

      使用python写MapReduce的“诀窍”是利用Hadoop流的API,通过STDIN(标准输入)、STDOUT(标准输出)在Map函数和Reduce函数之间传递数据。

      我们唯一需要做的是利用Python的sys.stdin读取输入数据,并把我们的输出传送给sys.stdout。Hadoop流将会帮助我们处理别的任何事情。

1.1 Map阶段:mapper.py

在这里,我们假设把文件保存到hadoop-0.20.2/test/code/mapper.py

#!/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)

    文件从STDIN读取文件。把单词切开,并把单词和词频输出STDOUT。Map脚本不会计算单词的总数,而是输出<word> 1。在我们的例子中,我们让随后的Reduce阶段做统计工作。

为了是脚本可执行,增加mapper.py的可执行权限

chmod +x hadoop-0.20.2/test/code/mapper.py

1.2 Reduce阶段:reducer.py

在这里,我们假设把文件保存到hadoop-0.20.2/test/code/reducer.py

#!/usr/bin/env python
from operator import itemgetter
import sys

current_word = None
current_count = 0
word = None

for line in sys.stdin:
    line = line.strip()
    word, count = line.split('\t', 1)
    try:
        count = int(count)
    except ValueError:  #count如果不是数字的话,直接忽略掉
        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)

文件会读取mapper.py 的结果作为reducer.py 的输入,并统计每个单词出现的总的次数,把最终的结果输出到STDOUT。

为了是脚本可执行,增加reducer.py的可执行权限

chmod +x hadoop-0.20.2/test/code/reducer.py

细节:split(chara, m),第二个参数的作用,下面的例子很给力

str = 'server=mpilgrim&ip=10.10.10.10&port=8080'
print str.split('=', 1)[0]  #1表示=只截一次
print str.split('=', 1)[1]
print str.split('=')[0]
print str.split('=')[1]

输出:

server
mpilgrim&ip=10.10.10.10&port=8080
server
mpilgrim&ip 

1.3 测试代码(cat data | map | sort | reduce)

这里建议大家在提交给MapReduce job之前在本地测试mapper.py 和reducer.py脚本。否则jobs可能会成功执行,但是结果并非自己想要的。

功能性测试mapper.py 和 reducer.py

[rte@hadoop-0.20.2]$cd test/code
[rte@code]$echo "foo foo quux labs foo bar quux" | ./mapper.py
foo 1
foo 1
quux    1
labs    1
foo 1
bar 1
quux    1
[rte@code]$echo "foo foo quux labs foo bar quux" | ./mapper.py | sort -k1,1 | ./reducer.py
bar 1
foo 3
labs    1
quux    2

细节:sort -k1,1  参数何意?

-k, -key=POS1[,POS2]     键以pos1开始,以pos2结束

有时候经常使用sort来排序,需要预处理把需要排序的field语言在最前面。实际上这是

完全没有必要的,利用-k参数就足够了。

比如sort all

1 4
2 3
3 2
4 1
5 0

如果sort -k 2的话,那么执行结果就是

5 0
4 1
3 2
2 3
1 4

2. 在Hadoop上运行python代码

2.1 数据准备

下载以下三个文件

我把上面三个文件放到hadoop-0.20.2/test/datas/目录下

2.2 运行

把本地的数据文件拷贝到分布式文件系统HDFS中。

bin/hadoop dfs -copyFromLocal /test/datas  hdfs_in

查看

bin/hadoop dfs -ls

结果

drwxr-xr-x   - rte supergroup          0 2014-07-05 15:40 /user/rte/hdfs_in

查看具体的文件

bin/hadoop dfs -ls /user/rte/hdfs_in

执行MapReduce job

bin/hadoop jar contrib/streaming/hadoop-*streaming*.jar \
-file test/code/mapper.py     -mapper test/code/mapper.py \
-file test/code/reducer.py    -reducer test/code/reducer.py \
-input /user/rte/hdfs_in/*    -output /user/rte/hdfs_out

实例输出

查看输出结果是否在目标目录/user/rte/hdfs_out

bin/hadoop dfs -ls /user/rte/hdfs_out

输出

Found 2 items
drwxr-xr-x   - rte supergroup          0 2014-07-05 20:51 /user/rte/hdfs_out2/_logs
-rw-r--r--   2 rte supergroup     880829 2014-07-05 20:51 /user/rte/hdfs_out2/part-00000

查看结果

bin/hadoop dfs -cat /user/rte/hdfs_out2/part-00000

输出

以上已经达成目的了,但是可以利用python迭代器和生成器优化

3. 利用python的迭代器和生成器优化Mapper 和 Reducer代码

3.1 python中的迭代器和生成器

   有关python的迭代器详情

3.2 优化Mapper 和 Reducer代码

mapper.py

#!/usr/bin/env python
import sys
def read_input(file):
    for line in file:
        yield line.split()

def main(separator='\t'):
    data = read_input(sys.stdin)
    for words in data:
        for word in words:
            print "%s%s%d" % (word, separator, 1)

if __name__ == "__main__":
    main()

reducer.py

#!/usr/bin/env python
from operator import itemgetter
from itertools import groupby
import sys

def read_mapper_output(file, separator = '\t'):
    for line in file:
        yield line.rstrip().split(separator, 1)

def main(separator = '\t'):
    data = read_mapper_output(sys.stdin, separator = separator)
    for current_word, group in groupby(data, itemgetter(0)):
        try:
            total_count = sum(int(count) for current_word, count in group)
            print "%s%s%d" % (current_word, separator, total_count)
        except valueError:
            pass

if __name__ == "__main__":
    main()

细节:groupby

from itertools import groupby
from operator import itemgetter

things = [('2009-09-02', 11),
          ('2009-09-02', 3),
          ('2009-09-03', 10),
          ('2009-09-03', 4),
          ('2009-09-03', 22),
          ('2009-09-06', 33)]

sss = groupby(things, itemgetter(0))
for key, items in sss:
    print key
    for subitem in items:
        print subitem
    print '-' * 20

结果

>>>
2009-09-02
('2009-09-02', 11)
('2009-09-02', 3)
--------------------
2009-09-03
('2009-09-03', 10)
('2009-09-03', 4)
('2009-09-03', 22)
--------------------
2009-09-06
('2009-09-06', 33)
--------------------

注 

  • groupby(things, itemgetter(0)) 以第0列为排序目标

  • groupby(things, itemgetter(1))以第1列为排序目标

  • groupby(things)以整行为排序目标



© 著作权归作者所有

共有 人打赏支持
潘少online
粉丝 11
博文 58
码字总数 107019
作品 2
深圳
程序员
使用Python实现Hadoop MapReduce程序

笔者的机器运行效果如下(输入数据是find的帮助手册,和笔者预期一样,the是最多的): --------------------------------------以下是原帖--------------------------------- 在这个实例中,...

彼得
2014/04/03
0
4
hadoop 里执行 MapReduce 任务的几种常见方式

说明: 测试文件: echo -e "aatbb tccnbbtcctdd" > 3.txt hadoop fs -put 3.txt /tmp/3.txt 全文的例子均以该文件做测试用例,统计单词出现的次数(WordCount)。 1、原生态的方式:java 源...

大数据之路
2012/10/14
0
0
Streaming执行Python版WordCount

一:先写map类 import sysfor line in sys.stdin:line = line.strip( )words = line.split( )for word in words:print('%st%s' % (word, 1)) 二:写reduce类 import syscurrent_word = Nonec......

白话
06/26
0
0
Python 的mapreduce 单词统计(转载)

!/usr/bin/env python import random 'abc..z' alphaStr = "".join(map(chr, range(97,123)))fp = open("word.txt", "w")maxIter = 100000for i in range(maxIter):word = ""len =random.ran......

MtrS
2014/12/23
0
0
Linux下使用Eclipse开发Hadoop应用程序 测试成功

参考:http://blog.sina.com.cn/s/blog62186b4601012acs.html HDFS 测试文件: 测试结果: 详细配置参考它的原文: Hadoop版本:hadoop-0.20.2 Eclipse版本:eclipse-java-helios-SR2-linux...

杨泽
2012/07/13
0
0

没有更多内容

加载失败,请刷新页面

加载更多

如何通过 J2Cache 实现分布式 session 存储

做 Java Web 开发的人多数都会需要使用到 session (会话),我们使用 session 来保存一些需要在两个不同的请求之间共享数据。一般 Java 的 Web 容器像 Tomcat、Resin、Jetty 等等,它们会在...

红薯
38分钟前
1
0
C++ std::thread

C++11提供了std::thread类来表示一个多线程对象。 1,首先介绍一下std::this_thread命名空间: (1)std::this_thread::get_id():返回当前线程id (2)std::this_thread::yield():用户接口...

yepanl
今天
2
0
Nignx缓存文件与动态文件自动均衡的配置

下面这段nginx的配置脚本的作用是,自动判断是否存在缓存文件,如果有优先输出缓存文件,不经过php,如果没有,则回到php去处理,同时生成缓存文件。 PHP框架是ThinkPHP,最后一个rewrite有关...

swingcoder
今天
1
0
20180920 usermod命令与用户密码管理

命令 usermod usermod 命令的选项和 useradd 差不多。 一个用户可以属于多个组,但是gid只有一个;除了gid,其他的组(groups)叫做扩展组。 usermod -u 1010 username # 更改用户idusermod ...

野雪球
今天
1
0
Java网络编程基础

1. 简单了解网络通信协议TCP/IP网络模型相关名词 应用层(HTTP,FTP,DNS等) 传输层(TCP,UDP) 网络层(IP,ICMP等) 链路层(驱动程序,接口等) 链路层:用于定义物理传输通道,通常是对...

江左煤郎
今天
1
0

没有更多内容

加载失败,请刷新页面

加载更多

返回顶部
顶部