文档章节

使用Python实现Hadoop MapReduce程序

彼得
 彼得
发布于 2014/04/03 17:38
字数 1805
阅读 6.1K
收藏 12

笔者的机器运行效果如下(输入数据是find的帮助手册,和笔者预期一样,the是最多的):


--------------------------------------以下是原帖---------------------------------

  • 在这个实例中,我将会向大家介绍如何使用Python 为 Hadoop编写一个简单的MapReduce

    程序。
    尽管Hadoop 框架是使用Java编写的但是我们仍然需要使用像C++、Python等语言来实现 Hadoop程序。尽管Hadoop官方网站给的示例程序是使用Jython编写并打包成Jar文件,这样显然造成了不便,其实,不一定非要这样来实现,我们可以使用Python与Hadoop 关联进行编程,看看位于/src/examples/python/WordCount.py  的例子,你将了解到我在说什么。

    我们想要做什么?

    我们将编写一个简单的 MapReduce 程序,使用的是C-Python,而不是Jython编写后打包成jar包的程序。
    我们的这个例子将模仿 WordCount 并使用Python来实现,例子通过读取文本文件来统计出单词的出现次数。结果也以文本形式输出,每一行包含一个单词和单词出现的次数,两者中间使用制表符来想间隔。

    先决条件

    编写这个程序之前,你学要架设好Hadoop 集群,这样才能不会在后期工作抓瞎。如果你没有架设好,那么在后面有个简明教程来教你在Ubuntu Linux 上搭建(同样适用于其他发行版linux、unix)

    如何使用Hadoop Distributed File System (HDFS)在Ubuntu Linux 建立单节点的 Hadoop 集群

    如何使用Hadoop Distributed File System (HDFS)在Ubuntu Linux 建立多节点的 Hadoop 集群


    Python的MapReduce代码


    使用Python编写MapReduce代码的技巧就在于我们使用了 HadoopStreaming 来帮助我们在Map 和 Reduce间传递数据通过STDIN (标准输入)和STDOUT (标准输出).我们仅仅使用Python的sys.stdin来输入数据,使用sys.stdout输出数据,这样做是因为HadoopStreaming会帮我们办好其他事。这是真的,别不相信!

    Map: mapper.py


    将下列的代码保存在/home/hadoop/mapper.py中,他将从STDIN读取数据并将单词成行分隔开,生成一个列表映射单词与发生次数的关系:
    注意:要确保这个脚本有足够权限(chmod +x /home/hadoop/mapper.py)。

    #!/usr/bin/env python
     
    import sys
     
    # input comes from STDIN (standard input)
    for line in sys.stdin:
        # remove leading and trailing whitespace
        line = line.strip()
        # split the line into words
        words = line.split()
        # increase counters
        for word in words:
            # write the results to STDOUT (standard output);
            # what we output here will be the input for the
            # Reduce step, i.e. the input for reducer.py
            #
            # tab-delimited; the trivial word count is 1
            print '%s\\t%s' % (word, 1)

    在这个脚本中,并不计算出单词出现的总数,它将输出 "<word> 1" 迅速地,尽管<word>可能会在输入中出现多次,计算是留给后来的Reduce步骤(或叫做程序)来实现。当然你可以改变下编码风格,完全尊重你的习惯。

    Reduce: reducer.py


    将代码存储在/home/hadoop/reducer.py 中,这个脚本的作用是从mapper.py 的STDIN中读取结果,然后计算每个单词出现次数的总和,并输出结果到STDOUT。
    同样,要注意脚本权限:chmod +x /home/hadoop/reducer.py

    #!/usr/bin/env python
     
    from operator import itemgetter
    import sys
     
    # maps words to their counts
    word2count = {}
     
    # input comes from STDIN
    for line in sys.stdin:
        # remove leading and trailing whitespace
        line = line.strip()
     
        # parse the input we got from mapper.py
        word, count = line.split('\\t', 1)
        # convert count (currently a string) to int
        try:
            count = int(count)
            word2count[word] = word2count.get(word, 0) + count
        except ValueError:
            # count was not a number, so silently
            # ignore/discard this line
            pass
     
    # sort the words lexigraphically;
    #
    # this step is NOT required, we just do it so that our
    # final output will look more like the official Hadoop
    # word count examples
    sorted_word2count = sorted(word2count.items(), key=itemgetter(0))
     
    # write the results to STDOUT (standard output)
    for word, count in sorted_word2count:
        print '%s\\t%s'% (word, count)


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


    我建议你在运行MapReduce job测试前尝试手工测试你的mapper.py 和 reducer.py脚本,以免得不到任何返回结果
    这里有一些建议,关于如何测试你的Map和Reduce的功能:

    ——————————————————————————————————————————————

    \r\n

     # very basic test
     hadoop@ubuntu:~$ echo "foo foo quux labs foo bar quux" | /home/hadoop/mapper.py
     foo     1
     foo     1
     quux    1
     labs    1
     foo     1
     bar     1
    ——————————————————————————————————————————————
     hadoop@ubuntu:~$ echo "foo foo quux labs foo bar quux" | /home/hadoop/mapper.py | sort | /home/hadoop/reducer.py
     bar     1
     foo     3
     labs    1
    ——————————————————————————————————————————————
    
     # using on[object Object]e of the ebooks as example input
     # (see below on where to get the ebooks)
     hadoop@ubuntu:~$ cat /tmp/gutenberg/20417-8.txt | /home/hadoop/mapper.py
     The     1
     Project 1
     Gutenberg       1
     EBook   1
     of      1
     [...] 
     (you get the idea)
    
     quux    2
    
     quux    1


    ——————————————————————————————————————————————
    
    
    
    为了这个例子,我们将需要三种电子书:

    下载他们,并使用us-ascii编码存储 解压后的文件,保存在临时目录,比如/tmp/gutenberg.

     hadoop@ubuntu:~$ ls -l /tmp/gutenberg/
     total 3592
     -rw-r--r-- 1 hadoop hadoop  674425 2007-01-22 12:56 20417-8.txt
     -rw-r--r-- 1 hadoop hadoop 1423808 2006-08-03 16:36 7ldvc10.txt
     -rw-r--r-- 1 hadoop hadoop 1561677 2004-11-26 09:48 ulyss12.txt
     hadoop@ubuntu:~$
    在我们运行MapReduce job 前,我们需要将本地的文件复制到HDFS中:
    
     hadoop@ubuntu:/usr/local/hadoop$ bin/hadoop dfs -copyFromLocal /tmp/gutenberg gutenberg
     hadoop@ubuntu:/usr/local/hadoop$ bin/hadoop dfs -ls
     Found 1 items
     /user/hadoop/gutenberg  <dir>
     hadoop@ubuntu:/usr/local/hadoop$ bin/hadoop dfs -ls gutenberg
     Found 3 items
     /user/hadoop/gutenberg/20417-8.txt      <r 1>   674425
     /user/hadoop/gutenberg/7ldvc10.txt      <r 1>   1423808
     /user/hadoop/gutenberg/ulyss12.txt      <r 1>   1561677
    
    
    
    现在,一切准备就绪,我们将在运行Python MapReduce job 在Hadoop集群上。像我上面所说的,我们使用的是
     帮助我们传递数据在Map和Reduce间并通过STDIN和STDOUT,进行标准化输入输出。
    
     
     hadoop@ubuntu:/usr/local/hadoop$ bin/hadoop jar contrib/streaming/hadoop-0.19.1-streaming.jar
     -mapper /home/hadoop/mapper.py -reducer /home/hadoop/reducer.py -input gutenberg/* 
    -output gutenberg-output
    在运行中,如果你想更改Hadoop的一些设置,如增加Reduce任务的数量,你可以使用“-hadoop@ubuntu:/usr/local/hadoop$ bin/hadoop jar contrib/streaming/hadoop-0.19.1-streaming.jar 
     -mapper ...
    
    一个重要的备忘是关于 
    这个任务将会读取HDFS目录下的HDFS目录下的
    目录。
    之前执行的结果如下:
    
    hadoop@ubuntu:/usr/local/hadoop$ bin/hadoop jar contrib/streaming/hadoop-0.19.1-streaming.jar 
    -mapper /home/hadoop/mapper.py -reducer /home/hadoop/reducer.py -input gutenberg/* 
    -output gutenberg-output
     
    additionalConfSpec_:null
     null=@@@userJobConfProps_.get(stream.shipped.hadoopstreaming
     packageJobJar: [/usr/local/hadoop-datastore/hadoop-hadoop/hadoop-unjar54543/]
     [] /tmp/streamjob54544.jar tmpDir=null
     [...] INFO mapred.FileInputFormat: Total input paths to process : 7
     [...] INFO streaming.StreamJob: getLocalDirs(): [/usr/local/hadoop-datastore/hadoop-hadoop/mapred/local]
     [...] INFO streaming.StreamJob: Running job: job_200803031615_0021
     [...]
     [...] INFO streaming.StreamJob:  map 0%  reduce 0%
     [...] INFO streaming.StreamJob:  map 43%  reduce 0%
     [...] INFO streaming.StreamJob:  map 86%  reduce 0%
     [...] INFO streaming.StreamJob:  map 100%  reduce 0%
     [...] INFO streaming.StreamJob:  map 100%  reduce 33%
     [...] INFO streaming.StreamJob:  map 100%  reduce 70%
     [...] INFO streaming.StreamJob:  map 100%  reduce 77%
     [...] INFO streaming.StreamJob:  map 100%  reduce 100%
     [...] INFO streaming.StreamJob: Job complete: job_200803031615_0021
    
    
     [...] INFO streaming.StreamJob: Output: gutenberg-output  hadoop@ubuntu:/usr/local/hadoop$ 
    
    
    正如你所见到的上面的输出结果,Hadoop 同时还提供了一个基本的WEB接口显示统计结果和信息。
    当Hadoop集群在执行时,你可以使用浏览器访问   ,如图:
    
    
    
    
    检查结果是否输出并存储在HDFS目录下的中:
    
     hadoop@ubuntu:/usr/local/hadoop$ bin/hadoop dfs -ls gutenberg-output
     Found 1 items
     /user/hadoop/gutenberg-output/part-00000     <r 1>   903193  2007-09-21 13:00
     hadoop@ubuntu:/usr/local/hadoop$ 
    
    可以使用 命令检查文件目录
    
     hadoop@ubuntu:/usr/local/hadoop$ bin/hadoop dfs -cat gutenberg-output/part-00000
     "(Lo)cra"       1
     "1490   1
     "1498," 1
     "35"    1
     "40,"   1
     "A      2
     "AS-IS".        2
     "A_     1
     "Absoluti       1
     [...]
     hadoop@ubuntu:/usr/local/hadoop$
    
    注意比输出,上面结果的(")符号不是Hadoop插入的。
    
    
     
    
     请参考:
    
    
    http://www.michael-noll.com/wiki/Writing_An_Hadoop_MapReduce_Program_In_Python#What_we_want_to_do



本文转载自:http://asfr.blogbus.com/logs/44208067.html

彼得

彼得

粉丝 39
博文 122
码字总数 38726
作品 0
深圳
程序员
私信 提问
加载中

评论(4)

毕竟杰神
毕竟杰神
java version "1.8.0_65"
Java(TM) SE Runtime Environment (build 1.8.0_65-b17)
Java HotSpot(TM) 64-Bit Server VM (build 25.65-b01, mixed mode)
woshifuwa
woshifuwa
支持一下
彼得
彼得 博主
java version "1.7.0_51"
Java(TM) SE Runtime Environment (build 1.7.0_51-b13)
Java HotSpot(TM) 64-Bit Server VM (build 24.51-b03, mixed mode)
彼得
彼得 博主
Linux linux-peterguo 2.6.31.5-0.1-desktop #1 SMP PREEMPT 2009-10-26 15:49:03 +0100 x86_64 x86_64 x86_64 GNU/Linux
openSUSE 11.2 (x86_64)
VERSION = 11.
java version "1.6.0_20"
OpenJDK Runtime Environment (IcedTea6 1.9.7) (suse-1.2.2-x86_64)
OpenJDK 64-Bit Server VM (build 19.0-b09, mixed mode)
javac 1.6.0_20
Pydoop 架构和模块包介绍

背景 Hadoop支持Java以及C/C++开发,其中Hadoop Pipes(C++)支持MapReduce,而Hadoop libhdfs(C)支持HDFS的访问。但Python的解决方案仅能支持Hadoop部分功能,具体如下。 Hadoop Streaming:1...

索隆
2012/09/13
447
0
Hadoop MapReduce计算框架

1、MapReduce理论 1.1、MapReduce是什么? MapReduce用于处理海量数据的分布式计算框架,是Hadoop生态中的核心之一(MapReduce用于计算海量数据,HDFS用于存储海量数据);MapReduce是谷歌公...

巴利奇
2018/10/30
0
0
【大数据分析常用算法】0.简介

简介 本系列教程主要探讨有关大数据的所有常用算法的实现,定期加入有趣的项目实践。 1、实现语言 初步使用java实现,后续加入python。 Hadoop Mapreduce Java实现 位于包*.mapreduce下。 Sp...

Areya
2019/01/14
110
0
大数据技术学习,大数据处理为何选择Spark,而不是Hadoop

大数据处理为何选择Spark,而不是Hadoop。 一.基础知识 1.Spark Spark是一个用来实现快速而通用的集群计算的平台。 在速度方面,Spark扩展了广泛使用的MapReduce计算模型,而且高效地支持更多...

董黎明
2018/10/20
37
0
【电子书】Hadoop实战手册 (样章第一章)

Hadoop实战手册 [美] Jonathan R. Owens,Jon Lentz,Brian Femiano 著; 傅杰,赵磊,卢学裕 译 内容简介   这是一本Hadoop实用手册,主要针对实际问题给出相应的解决方案。《Hadoop实战手...

dwf07223
2018/06/28
0
0

没有更多内容

加载失败,请刷新页面

加载更多

每天AC系列(一):三数之和

1 题目 LeetCode第15题,难度中等,题目描述: 给定一个包含 n 个整数的数组 nums,判断 nums 中是否存在三个元素 a,b,c ,使得 a + b + c = 0 ?找出所有满足条件且不重复的三元组。 注意:答...

Blueeeeeee
26分钟前
56
0
OSChina 周四乱弹 —— 水果你们都没吃全

Osc乱弹歌单(2020)请戳(这里) 【今日歌曲】 @ 莱布妮子:分享五月天的单曲《温柔》@小小编辑 @cIouddyy @clouddyy 《温柔》- 五月天 手机党少年们想听歌,请使劲儿戳(这里) @FalconChe...

小小编辑
44分钟前
91
1
聚合支付网站被黑客攻击 导致数据库被篡改的防御办法

2020春节即将来临,收到新聚合支付平台网站客户的求助电话给我们Sinesafe,反映支付订单状态被修改由原先未支付修改为已支付,导致商户那边直接发货给此订单会员了,商户和平台的损失较大,很多码...

网站安全
昨天
63
0
MySQL-基于SELECT查询的UPDATE查询

我需要检查(从同一张表)基于日期时间的两个事件之间是否存在关联。 一组数据将包含某些事件的结束日期时间,另一组数据将包含其他事件的开始日期时间。 如果第一个事件在第二个事件之前完成...

javail
昨天
70
0
将PostgreSQL数据库复制到另一台服务器

我正在将生产PostgreSQL数据库复制到开发服务器。 什么是最快,最简单的方法? #1楼 pg_dump the_db_name > the_backup.sql 然后将备份复制到您的开发服务器,并使用以下命令进行还原: ps...

技术盛宴
昨天
130
0

没有更多内容

加载失败,请刷新页面

加载更多

返回顶部
顶部