Hadoop streaming

Hadoop為MapReduce提供了不同的API，可以方便我們使用不同的編程語言來使用MapReduce框架，而不是只局限于Java。這里要介紹的就是Hadoop streaming API。Hadoop streaming 使用Unix的standard streams作為我們mapreduce程序和MapReduce框架之間的接口。所以你可以用任何語言來編寫MapReduce程序，只要該語言可以往standard input/output上進行讀寫。

streamming是天然適用于文字處理的（text processing），當然，也僅適用純文本的處理，對于需要對象和序列化的場景，hadoop streaming無能為力。它力圖使我們能夠快捷的通過各種腳本語言，快速的處理大量的文本文件。以下是steaming的一些特點：

常用的Streaming編程語言：

Ruby

下面是一個Ruby編寫的MapReduce程序的示例：

map

max_temperature_map.rb：

ruby #!/usr/bin/env ruby STDIN.each_line do |line| val = line year, temp, q = val[15,4], val[87,5], val[92,1] puts "#{year}/t#{temp}" if (temp != "+9999" && q =~ /[01459]/) end 

reduce

max_temperature_reduce.rb：

ruby #!/usr/bin/env ruby last_key, max_val = nil, -1000000 STDIN.each_line do |line| key, val = line.split("/t") if last_key && last_key != key puts "#{last_key}/t#{max_val}" last_key, max_val = key, val.to_i else last_key, max_val = key, [max_val, val.to_i].max end end puts "#{last_key}/t#{max_val}" if last_key 

運行

% 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

Python

Map

#!/usr/bin/env pythonimport reimport sysfor line in sys.stdin:val = line.strip()(year, temp, q) = (val[15:19], val[87:92], val[92:93])if (temp != "+9999" and re.match("[01459]", q)):print "%s/t%s" % (year, temp)

Reduce

#!/usr/bin/env pythonimport sys(last_key, max_val) = (None, -sys.maxint)for line in sys.stdin:(key, val) = line.strip().split("/t")if last_key and last_key != key:print "%s/t%s" % (last_key, max_val)(last_key, max_val) = (key, int(val))else:(last_key, max_val) = (key, max(max_val, int(val)))if last_key:print "%s/t%s" % (last_key, max_val)