快捷導(dǎo)航

Hadoop streaming詳細(xì)介紹

更新時(shí)間：2017年03月15日 16:19:27 作者：Hadoop streaming

這篇文章主要介紹了Hadoop streaming詳細(xì)介紹的相關(guān)資料,需要的朋友可以參考下

Hadoop streaming

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

streamming是天然適用于文字處理的（text processing），當(dāng)然，也僅適用純文本的處理，對(duì)于需要對(duì)象和序列化的場(chǎng)景，hadoop streaming無(wú)能為力。它力圖使我們能夠快捷的通過(guò)各種腳本語(yǔ)言，快速的處理大量的文本文件。以下是steaming的一些特點(diǎn)：

Map函數(shù)的輸入是通過(guò)stand input一行一行的接收數(shù)據(jù)的。（不像Java API，通過(guò)InputFormat類(lèi)做預(yù)處理，使得Map函數(shù)的輸入是有Key和value的）
Map函數(shù)的output則必須限定為key-value pair，key和value之間用\t分開(kāi)。（MapReduce框架在處理intermediate的Map輸出時(shí)，必須做sort和partition，即shuffle）
Reduce函數(shù)的input是Map函數(shù)的output也是key-value pair，key和value之間用\t分開(kāi)。

常用的Streaming編程語(yǔ)言：

bash shell
ruby
python

Ruby

下面是一個(gè)Ruby編寫(xiě)的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

從標(biāo)準(zhǔn)輸入讀入一行data。
處理數(shù)據(jù)之后，生成一個(gè)鍵值對(duì)，用\t分隔，輸出到標(biāo)準(zhǔn)輸出

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

從標(biāo)準(zhǔn)輸入讀入一行數(shù)據(jù)
數(shù)據(jù)是用\t分隔的鍵值對(duì)
數(shù)據(jù)是被MapReduce根據(jù)key排序之后順序一行一行讀入
reduce函數(shù)對(duì)數(shù)據(jù)進(jìn)行處理，并輸出，輸出仍是用\t分隔的鍵值對(duì)

運(yùn)行

% 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

hadoop jar $HADOOP_INSTALL/contrib/streaming/hadoop-*-streaming.jar指明了使用hadoop streaming
hadoop-*-streaming.jar會(huì)將input里的文件，一行一行的輸出到標(biāo)準(zhǔn)輸出。
用-mapper指定Map函數(shù)。類(lèi)似于通過(guò)管道將數(shù)據(jù)傳給rb文件： data|ch02/src/main/ruby/max_temperature_map.rb
-reducer指定Reduce函數(shù)。

Python

Map

#!/usr/bin/env python
import re
import sys
for 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 python
import 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)

運(yùn)行

% hadoop jar $HADOOP_INSTALL/contrib/streaming/hadoop-*-streaming.jar \
-input input/ncdc/sample.txt \
-output output \
-mapper ch02/src/main/ruby/max_temperature_map.py\
-reducer ch02/src/main/ruby/max_temperature_reduce.py

Bash shell

Map

#!/usr/bin/env bash
# NLineInputFormat gives a single line: key is offset, value is S3 URI
read offset s3file
# Retrieve file from S3 to local disk
echo "reporter:status:Retrieving $s3file" >&2
$HADOOP_INSTALL/bin/hadoop fs -get $s3file .
# Un-bzip and un-tar the local file
target=`basename $s3file .tar.bz2`
mkdir -p $target
echo "reporter:status:Un-tarring $s3file to $target" >&2
tar jxf `basename $s3file` -C $target
# Un-gzip each station file and concat into one file
echo "reporter:status:Un-gzipping $target" >&2
for file in $target/*/*
do
gunzip -c $file >> $target.all
echo "reporter:status:Processed $file" >&2
done
# Put gzipped version into HDFS
echo "reporter:status:Gzipping $target and putting in HDFS" >&2
gzip -c $target.all | $HADOOP_INSTALL/bin/hadoop fs -put - gz/$target.gz

運(yùn)行

% hadoop jar $HADOOP_INSTALL/contrib/streaming/hadoop-*-streaming.jar \
-D mapred.reduce.tasks=0 \
-D mapred.map.tasks.speculative.execution=false \
-D mapred.task.timeout=12000000 \
-input ncdc_files.txt \
-inputformat org.apache.hadoop.mapred.lib.NLineInputFormat \
-output output \
-mapper load_ncdc_map.sh \
-file load_ncdc_map.sh

這里的-D mapred.reduce.tasks=0將reduce task觀(guān)掉，因此也不需要設(shè)置-reducer
只使用Mapper，可以通過(guò)MapReduce幫助我們并行的完成一些平時(shí)只能串行的shell腳本
注意這里的-file，在集群模式下，需要并行運(yùn)行時(shí)，需要-file把文件傳輸?shù)狡渌?jié)點(diǎn)

Combiner

在streaming模式下，仍然可以運(yùn)行Combiner，兩種方法：

通過(guò)Java編寫(xiě)一個(gè)combiner的函數(shù)，并使用-combiner option
以命令行的管道模式完成combiner的任務(wù)

這里具體解釋第二種方法：

% 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

注意看-mapper這一行，通關(guān)管道的方式，把mapper的臨時(shí)輸出文件（intermediate file，Map完成后的臨時(shí)文件）作為輸入，送到sort進(jìn)行排序，然后送到reduce腳本，來(lái)完成類(lèi)似于combiner的工作。這時(shí)候的輸出才真正的作為shuffle的輸入，被分組并在網(wǎng)絡(luò)上發(fā)送到Reduce

感謝閱讀，希望能幫助到大家，謝謝大家對(duì)本站的支持！

您可能感興趣的文章: