Hadoop中的streaming
Hadoop为MapReduce提供了一个允许你使用除了java以外的语言编写map，reduce函数的API：Hadoop Streaming使用标准流（standard streams)作为Hadoop和应用程序之间传输数据的接口。所以你可以使用任何语言编写map，reduce函数，只要它能够从标准输入流（stdin）中读入数据，以及向标准输出流（stdout）中写输出数据就行。
Streaming本质上就非常适合处理文本数据。Map输入数据通过stdin传输到map函数中，map函数再一行一行的处理这些输入数据，然后被处理过的以tab分割的键值对被写到stdout中去，reduce函数从stdin中接收到这些map输出（Hadoop框架保证这些map输出是经过关键字排序的），然后把处理后的reduce输出写到stdout中。
以ruby为例
以 学习笔记（一） 中的计算MaxTemperature的程序为例，其Map函数的ruby代码表示如下（即使你像我一样没接触过ruby，你一样可以读懂下面这段代码）：
1 # !/usr/bin/env ruby
2 STDIN.each_line do |line|
3 val = line
4 year, temp, q = val[15,4], val[87,5], val[92,1 ]
5 puts " #{year}\t#{temp} " if (temp != " +9999 " && q =~ /[01459]/ )
6 end
这个程序为从stdin中读入的每行数据执行此代码块。这段代码块从每行中抽取出相关的字段（年份，温度和质量），如果温度值有效，就把年份和温度值写入到stdout中去（以tab键分隔开）。
因为该脚本只是在标准输入和输出上运行，所以最简单的方式就是在unix pipe上进行测试，而不是使用hadoop：
% cat input/ncdc/sample.txt | ch02/src/main/ruby/max_temperature_map.rb1950 +0000
1950 +0022
1950 -0011
1949 +0111
1949 +0078
相应的，ruby形式的reduce函数如下所示：
1 # !/usr/bin/env ruby
2 last_key, max_val = nil, -1000000
3 STDIN.each_line do |line|
4 key, val = line.split( " \t " )
5 if last_key && last_key != key
6 puts " #{last_key}\t#{max_val} "
7 last_key, max_val = key, val.to_i
8 else
9 last_key, max_val = key, [max_val, val.to_i].max
10 end
11 end
12 puts " #{last_key}\t#{max_val} " if last_key
同样的，这段代码迭代从stdin中读入的每一行，但是我们必须为当前正在处理的每一个key group保存一些数据。在这个例子中，key是年份值，我们把上一次处理过的年份值last_key和迄今为止对应的该key group中最大的温度值max_val存储下来。MapReduce框架会确保这些key值是经过排序的，所以当我们遇到一个不同的key值时，说明上一个key group已经处理完了，所保存的最大温度值max_val就是该key group最大的温度值。对比一下java实现，我们可以发现，java API会自动的为你分组，而在ruby中你不得不自己确定key group的界限。
对于每一行，我们抽取出年份和温度值，如果我们已经处理完了一组key-value pair（last_key && last_key != key），我们就把last_key和max_value（仍然以tab键分开）写入到stdout中去，然后重新设置last_key,max_value为对应的新值。如果还没有处理完某个key group，我们就只是更新下当前last_key所对应的max_value。现在我们仍然用unix pipe来模拟下整个mapreduce过程：
% cat input/ncdc/sample.txt | ch02/src/main/ruby/max_temperature_map.rb | \sort | ch02/src/main/ruby/max_temperature_reduce.rb
1949 111
1950 22
可以看到，这次输出跟java对应的输出是一样的，现在我们使用Hadoop来运行一下。由于hadoop命令不支持streaming选项，所以你必须通过jar选项声明要处理streaming 的streaming jar文件。如下所示：
% hadoop jar $HADOOP_INSTALL/contrib/streaming/hadoop-*-streaming.jar \-input input/ncdc/sample.txt \
-output output \
-mapper ch02/src/main/ruby/max_temperature_map.rb \
-reducer ch02/src/main/ruby/max_temperature_reduce.rb
如果是在一个较大的集群上运行改程序，我们可以使用-combiner选项来声明一个combiner（关于combiner的内容请参看 学习笔记（三） ）。早前的版本combiner可能限定于java代码实现，但在1.x以后的版本中，combiner可以是任意的streaming命令。如下所示：
% hadoop jar $HADOOP_INSTALL/contrib/streaming/hadoop-*-streaming.jar \-input input/ncdc/all \
-output output \
-mapper "ch02/src/main/ruby/max_temperature_map.rb | sort |
ch02/src/main/ruby/max_temperature_reduce.rb" \
-reducer ch02/src/main/ruby/max_temperature_reduce.rb \
-file ch02/src/main/ruby/max_temperature_map.rb \
-file ch02/src/main/ruby/max_temperature_reduce.rb
在本例中，我们指定输入文件为一个目录（-input input/ncdc/all），这样就把该目录下的所有文件读入mapper中了（在本例中此目录下是两个.gz文件，也就是gzip的压缩文件：1901.gz,1902.gz）。其中-file选项用来说明要拷贝到集群上的文件（如果是在单例模式下运行就不需要这么做了）。值得注意的是，此例中的-mapper选项同时指明了mapper和combiner（也就是max_temperature_reduce.rb，还记得吗，在 学习笔记（三） 中我们知道java代码的combiner跟reducer是同一实现）
以Python为例
熟悉python的同学可以参考一下python下map和reduce函数的代码：：
1 # !/usr/bin/env python 2 import re 3 import sys 4 for line in sys.stdin: 5 val = line.strip() 6 (year, temp, q) = (val[15:19], val[87:92], val[92:93 ]) 7 if (temp != " +9999 " and re.match( " [01459] " , q)): 8 print " %s\t%s " % (year, temp)
1 # !/usr/bin/env python 2 import sys 3 (last_key, max_val) = (None, - sys.maxint) 4 for line in sys.stdin: 5 (key, val) = line.strip().split( " \t " ) 6 if last_key and last_key != key: 7 print " %s\t%s " % (last_key, max_val) 8 (last_key, max_val) = (key, int(val)) 9 else : 10 (last_key, max_val) = (key, max(max_val, int(val))) 11 if last_key: 12 print " %s\t%s " % (last_key, max_val)
执行测试脚本命令为：
% cat input/ncdc/sample.txt | ch02/src/main/python/max_temperature_map.py | \sort | ch02/src/main/python/max_temperature_reduce.py
1949 111
1950 22
转载请注明出处： http://HdhCmsTestcnblogs测试数据/beanmoon/archive/2012/12/07/2807759.html
发散你的思维； stay hungry, stay foolish
分类: hadoop&大数据
标签: hadoop , combiner , streaming , ruby
作者： Leo_wl
出处： http://HdhCmsTestcnblogs测试数据/Leo_wl/
本文版权归作者和博客园共有，欢迎转载，但未经作者同意必须保留此段声明，且在文章页面明显位置给出原文连接，否则保留追究法律责任的权利。
版权信息查看更多关于Hadoop中的streaming的详细内容...