一、原理
Hadoop Streaming是Hadoop提供的一個編程工具，它允許用戶使用任何可執行檔案或者腳本檔案作為Mapper和Reducer，例如：采用shell腳本語言中的一些命令作為mapper和reducer（cat作為mapper，wc作為reducer）
$HADOOP_HOME/bin/hadoop jar $HADOOP_HOME/contrib/streaming/hadoop-*-streaming.jar \
-input myInputDirs \
-output myOutputDir \
-mapper cat \
-reducer wc
mapper和reducer會從標準輸入中讀取用戶資料，一行一行處理后發送給標準輸出。Streaming工具會創建MapReduce作業，發送給各個tasktracker，同時監控整個作業的執行程序。
如果一個檔案（可執行或者腳本）作為mapper，則在mapper初始化時，每一個mapper任務會把該檔案作為一個單獨行程啟動，mapper任務運行時，它把輸入切分成行并把每一行提供給可執行檔案行程的標準輸入。 同時，mapper收集可執行檔案行程標準輸出的內容，并把收到的每一行內容轉化成key/value對，作為mapper的輸出。默認情況下，一行中第一個tab之前的部分作為key，之后的（不包括tab）作為value。如果沒有tab，整行作為key值，value值為null。不過，這可以定制，在下文中會介紹如何自定義key和value的切分方式。
對于reducer，類似。
以上是Map/Reduce框架和streaming mapper/reducer之間的基本通信協議。
二、語法
1、基本語法
Usage: $HADOOP_HOME/bin/hadoop jar \
$HADOOP_HOME/contrib/streaming/hadoop-*-streaming.jar [options]
options：
（1）-input：輸入檔案路徑
（2）-output：輸出檔案路徑
（3）-mapper：用戶自己寫的mapper程式，可以是可執行檔案或者腳本
（4）-reducer：用戶自己寫的reducer程式，可以是可執行檔案或者腳本
（5）-file：打包檔案到提交的作業中，可以是mapper或者reducer要用的輸入檔案，如組態檔，字典等。
（6）-partitioner：用戶自定義的partitioner程式
（7）-combiner：用戶自定義的combiner程式（必須用java實作）
（8）-D：作業的一些屬性（以前用的是-jonconf），具體有：
1）mapred.map.tasks：map task數目
2）mapred.reduce.tasks：reduce task數目
3）stream.map.input.field.separator/stream.map.output.field.separator：map task輸入/輸出資料的分隔符，默認均為\t。
4）stream.num.map.output.key.fields：指定map task輸出記錄中key所占的域數目
5）stream.reduce.input.field.separator/stream.reduce.output.field.separator：reduce task輸入/輸出資料的分隔符，默認均為\t。
6）stream.num.reduce.output.key.fields：指定reduce task輸出記錄中key所占的域數目。
有時只需要map函式處理輸入資料，這時只需把mapred.reduce.tasks設定為零，Map/Reduce框架就不會創建reducer任務，mapper任務的輸出就是整個作業的最終輸出。為了做到向下兼容，Hadoop Streaming也支持“-reduce None”選項，它與“-jobconf mapred.reduce.tasks=0”等價。
2、擴展語法
之前已經提到，當Map/Reduce框架從mapper的標準輸入讀取一行時，它把這一行切分為key/value對。在默認情況下，每行第一個tab符之前的部分作為key，之后的部分作為value（不包括tab符）。
但是，用戶也可以自定義，可以指定分隔符是其它字符而不是默認的tab符，或者指定在第n（n>=1）個分割符處分割而不是默認的第一個。例如：
$HADOOP_HOME/bin/hadoop jar $HADOOP_HOME/hadoop-streaming.jar \
-input myInputDirs \
-output myOutputDir \
-mapper org.apache.hadoop.mapred.lib.IdentityMapper \
-reducer org.apache.hadoop.mapred.lib.IdentityReducer \
-jobconf stream.map.output.field.separator=. \
-jobconf stream.num.map.output.key.fields=4
在上面的例子中，“-jobconf stream.map.output.field.separator=.”指定“.”作為map輸出內容的分隔符，并且從在第4個“.”之前的部分作為key，之后的部分作為value（不包括這第4個“.”）。 如果一行中的“.”少于4個，則整行的內容作為key，value設為空的Text物件（就像這樣創建了一個Text：new Text("")）。
同樣地，用戶也可以使用“-jobconf stream.reduce.output.field.separator=SEP”和“-jobconf stream.num.reduce.output.fields=NUM”來指定reduce輸出的行中，第幾個分隔符處分割key和value。
三、實體
為了說明各種語言撰寫Hadoop Streaming程式的方法，下面以WordCount為例，WordCount作業的主要功能是對用戶輸入的資料中所有字串進行計數。
1、shell
#vi mapper.sh
#! /bin/bash
while read LINE; do
for word in $LINE
do
echo "$word 1"
done
done
-------------------------------------------------------------------------
#vi reducer.sh
#! /bin/bash
count=0
started=0
word=""
while read LINE;do
newword=`echo $LINE | cut -d ' ' -f 1`
if [ "$word" != "$newword" ];then
[ $started -ne 0 ] && echo -e "$word\t$count"
word=$newword
count=1
started=1
else
count=$(( $count + 1 ))
fi
done
echo -e "$word\t$count"
-------------------------------------------------------------------------
本地測驗：cat input.txt | sh mapper.sh | sort | sh reducer.sh
集群測驗：
$HADOOP_HOME/bin/hadoop jar $HADOOP_HOME/contrib/streaming/hadoop-*-streaming.jar \
-input myInputDirs \
-output myOutputDir \
-mapper mapper.sh\
-reducer reducer.sh
如果執行上面腳本提示：“Caused by: java.io.IOException: Cannot run program “/user/hadoop/Mapper”: error=2, No such file or directory”，則說明找不到可執行程式，可以在提交作業時，采用-file選項指定這些檔案，比如上面例子中，可以使用“-file mapper.py -file reducer.py”，這樣，Hadoop會將這兩個檔案自動分發到各個節點上，比如：
$HADOOP_HOME/bin/hadoop jar $HADOOP_HOME/contrib/streaming/hadoop-*-streaming.jar \
-input myInputDirs \
-output myOutputDir \
-mapper mapper.sh\
-reducer reducer.sh\
-file mapper.sh \
-file reducer.sh
2、python
#vi mapper.py
#!/usr/bin/env python
import sys
#maps words to their counts
word2count = {}
#input comes from STDIN (standard input)
for line in sys.stdin:
#remove leading and trailing whitespace
line = line.strip()
#split the line into words while removing any empty strings
words = filter(lambda word: word, 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)
-------------------------------------------------------------------------
#vi 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()
#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 input.txt | python mapper.py | sort | python reducer.py
集群測驗：
$HADOOP_HOME/bin/hadoop jar $HADOOP_HOME/contrib/streaming/hadoop-*-streaming.jar \
-input myInputDirs \
-output myOutputDir \
-mapper mapper.py\
-reducer reducer.py
uj5u.com熱心網友回復：
good 謝謝分享~~uj5u.com熱心網友回復：
看看 還不錯哦轉載請註明出處，本文鏈接：https://www.uj5u.com/qita/101539.html
標籤：華為云計算