使用Python实现Hadoop MapReduce程序

时间  2019-11-09
标签 使用 python 实现 hadoop mapreduce 程序 栏目 Python 繁體版
原文   原文链接

笔者的机器运行效果以下（输入数据是find的帮助手册，和笔者预期同样，the是最多的）：html


--------------------------------------如下是原帖---------------------------------python

  • 在这个实例中，我将会向你们介绍如何使用Python 为 Hadoop编写一个简单的MapReduce
    linux

    程序。
    尽管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会帮咱们办好其余事。这是真的，别不相信！

    apache

    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步骤（或叫作程序）来实现。固然你能够改变下编码风格，彻底尊重你的习惯。

    ubuntu

    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的功能：
    app

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

    \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.

    ide

     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
相关文章
相关标签/搜索
每日一句
    每一个你不满意的现在，都有一个你没有努力的曾经。
本站公众号
   欢迎关注本站公众号,获取更多信息
联系我们 最近搜索 最新文章 沪ICP备13005482号-10 MyBatis教程 SQL 教程 MySQL教程 Java 教程 Thymeleaf 教程 Hibernate教程 Spring教程 Redis教程