使用python编写hadoop的mapper和reducer的实例详解
hadoop streaming 原理
hadoop 本身是用 java 开发的,程序也需要用 java 编写,但是通过 hadoop streaming,我们可以使用任意语言来编写程序,让 hadoop 运行。
hadoop streaming 就是通过将其他语言编写的 mapper 和 reducer 通过参数传给一个事先写好的 java 程序(hadoop 自带的 *-streaming.jar),这个 java 程序会负责创建 mr 作业,另开一个进程来运行 mapper,将得到的输入通过 stdin 传给它,再将 mapper 处理后输出到 stdout 的数据交给 hadoop,经过 partition 和 sort 之后,再另开进程运行 reducer,同样通过 stdin/stdout 得到最终结果。
python的mapreduce代码
因此,使用python编写mapreduce代码的技巧就在于我们使用了 hadoopstreaming 来帮助我们在map 和 reduce间传递数据通过stdin (标准输入)和stdout (标准输出).我们仅仅使用python的sys.stdin来输入数据,使用sys.stdout输出数据,这样做是因为hadoopstreaming会帮我们办好其他事。
创建文件,上传文件
当前路径下,创建一本,包含英文单词(后面mapper 和reduce 统计单词频次需要使用)
hadoop@derekubun:/usr/local/hadoop$ bin/hdfs dfs -mkdir -p /input hadoop@derekubun:/usr/local/hadoop$ bin/hdfs dfs -put ./book.txt /input
编写mapper.py 文件
将下列的代码保存在/home/hadoop/example/mapper.py中,他将从stdin读取数据并将单词成行分隔开,生成一个列表映射单词与发生次数的关系:
注意:要确保这个脚本有足够权限(chmod +x 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)
编写reducer 文件
将代码存储在/home/hadoop/example/reducer.py 中,这个脚本的作用是从mapper.py 的stdout中读取结果,然后计算每个单词出现次数的总和,并输出结果到stdout。
同样,要注意脚本权限:chmod +x reducer.py
#!/usr/bin/env python from operator import itemgetter import sys current_word = none current_count = 0 word = none # 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) except valueerror: # count was not a number, so silently # ignore/discard this line continue # this if-switch only works because hadoop sorts map output # by key (here: word) before it is passed to the reducer if current_word == word: current_count += count else: if current_word: # write result to stdout print '%s\t%s' % (current_word, current_count) current_count = count current_word = word # do not forget to output the last word if needed! if current_word == word: print '%s\t%s' % (current_word, current_count)
自测
在运行mapreduce job测试前尝试手工测试你的mapper.py 和 reducer.py脚本,以免得不到任何返回结果。这里有一些建议,关于如何测试你的map和reduce的功能:
hadoop@derekubun:/usr/local/hadoop$ echo "foo foo quux labs foo bar quux" | ./mapper.py foo 1 foo 1 quux 1 labs 1 foo 1 bar 1 quux 1 hadoop@derekubun:/usr/local/hadoop$ echo "foo foo quux labs foo bar quux" |./mapper.py | sort |./reducer.py bar 1 foo 3 labs 1 quux 2
hadoop 运行
一切准备就绪,我们将在运行python mapreduce job 在hadoop集群上。像我上面所说的,我们使用的是hadoopstreaming 帮助我们传递数据在map和reduce间并通过stdin和stdout,进行标准化输入输出。
hadoop@derekubun:/usr/local/hadoop$ hadoop jar share/hadoop/tools/lib/hadoop-streaming-2.7.6.jar \ -mapper 'python mapper.py' \ -file /home/hadoop/example/mapper.py \ -reducer 'python reducer.py' \ -file /home/hadoop/example/reducer.py -input hdfs:/input/book.txt \ -output output
第一行是告诉 hadoop 运行 streaming 的 java 程序,接下来的是参数:
这里的mapper 后面跟的其实是一个命令。也就是说,-mapper 和 -reducer 后面跟的文件名不需要带上路径。而 -file 后的参数需要带上路径,为了让 hadoop 将程序分发给其他机器,需要-file 参数指明要分发的程序放在哪里。
注意:如果你在 mapper 后的命令用了引号,加上路径名反而会报错说找不到这个程序。(因为 -file 选项会将对应的本地参数文件上传至 hadoop streaming 的工作路径下,所以再执行 -mapper 对应的参数命令能直接找到对应的文件。
-input 和 -output 后面跟的是 hdfs 上的路径名,这里的 input/book.txt 指的是input 文件夹下刚才上传的文本文件,注意 -output 后面跟着的需要是一个不存在于 hdfs 上的路径,在产生输出的时候 hadoop 会帮你创建这个文件夹,如果已经存在的话就会产生冲突。因此每次执行 hadoop streaming 前可以通过脚本命令 hadoop fs -rmr 清除输出路径。
由于 mapper 和 reducer 参数跟的实际上是命令,所以如果每台机器上 python 的环境配置不一样的话,会用每台机器自己的配置去执行 python 程序。
结果获取
如果运行中遇到问题,注意看报错,然后进行调整。
运行结束之后,结果存储在hdfs上 output目录下。
查看结果:hadoop@derekubun:/usr/local/hadoop$ bin/hdfs dfs -ls output
从hdfs拷贝至本地:hadoop@derekubun:/usr/local/hadoop$ bin/hdfs dfs -get output/* ./
注:如果结果中包含_success 则说明本次运行成功。
上一篇: JS拼接HTML页面的方法教程