Hadoop數(shù)據(jù)流是Hadoop自帶發(fā)行的實(shí)用程序。該實(shí)用程序允許創(chuàng)建和運(yùn)行Map/Reduce任務(wù)的任何可執(zhí)行文件或腳本映射器和/或減速器。

使用Python示例

對于Hadoop的數(shù)據(jù)流，我們考慮的字計(jì)數(shù)問題。任何工作在Hadoop中必須有兩個階段：映射器和減速器。我們使用python腳本代碼映射器和減速器在Hadoop下運(yùn)行它。使用Perl和Ruby也是類似的。

映射階段代碼

!/usr/bin/python
import sys
# Input takes from standard input for myline in sys.stdin: 
# Remove whitespace either side myline = myline.strip() 
# Break the line into words words = myline.split() 
# Iterate the words list for myword in words: 
# Write the results to standard output print '%s\t%s' % (myword, 1)

請確保此文件具有執(zhí)行權(quán)限（使用chmod +x /home/ expert/hadoop-1.2.1/mapper.py）。

減速器階段代碼

#!/usr/bin/python
from operator import itemgetter 
import sys 
current_word = ""
current_count = 0 
word = "" 
# Input takes from standard input for myline in sys.stdin: 
# Remove whitespace either side myline = myline.strip() 
# Split the input we got from mapper.py word, count = myline.split('\t', 1) 
# Convert count variable to integer 
   try: 
      count = int(count) 
except ValueError: 
   # Count was not a number, so silently ignore this line continue
if current_word == word: 
   current_count += count 
else: 
   if current_word: 
      # Write result to standard output 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)

保存mapper.py和reducer.py 在 Hadoop 的主目錄映射器和減速器代碼。確保這些文件具有執(zhí)行權(quán)限（使用chmod +x mapper.py 和 chmod +x reducer.py）。由于python具有大小寫敏感，因此相同的代碼可以從以下鏈接下載。

wordCount程序的執(zhí)行

$ $HADOOP_HOME/bin/hadoop jar contrib/streaming/hadoop-streaming-1.
2.1.jar \
   -input input_dirs \ 
   -output output_dir \ 
   -mapper <path/mapper.py \ 
   -reducer <path/reducer.py

其中“\”用于續(xù)行以便于閱讀。

例如，

./bin/hadoop jar contrib/streaming/hadoop-streaming-1.2.1.jar -input myinput -output myoutput -mapper /home/expert/hadoop-1.2.1/mapper.py -reducer /home/expert/hadoop-1.2.1/reducer.py

數(shù)據(jù)流工作原理

在上面的例子中，這兩個映射器和減速是從標(biāo)準(zhǔn)輸入讀取作為輸入，并輸出到標(biāo)準(zhǔn)輸出到Python腳本。實(shí)用程序?qū)?chuàng)建一個Map/Reduce作業(yè)，并將作業(yè)提交到一個合適的集群，并監(jiān)督工作的進(jìn)展情況，直至完成。

當(dāng)指定映射器的腳本，每個映射任務(wù)將啟動腳本作為一個單獨(dú)的進(jìn)程時映射器初始化。作為mapper任務(wù)運(yùn)行時，輸入轉(zhuǎn)換成行給進(jìn)程的標(biāo)準(zhǔn)輸入（STDIN）。在此期間，映射器收集從該方法的標(biāo)準(zhǔn)輸出（stdout）面向行輸出和每一行轉(zhuǎn)換為鍵/值對，其被收集作為映射器的輸出。缺省情況下，一行到第一個制表符的前綴是鍵和行（不包括制表符）的其余部分為值。如果在該行沒有任何制表符，則整行鍵和值被視為null。然而，這可以被定制，每次需要1個。

當(dāng)指定減速腳本，每個減速器任務(wù)將啟動腳本作為一個單獨(dú)的進(jìn)程，然后減速初始化。減速器任務(wù)運(yùn)行時將其轉(zhuǎn)換其輸入鍵/值對，進(jìn)入行并將該行進(jìn)程的標(biāo)準(zhǔn)輸入（STDIN）。在此期間，在減速機(jī)收集來自該過程的標(biāo)準(zhǔn)輸出（stdout）的面向行的輸出，每行轉(zhuǎn)換成一個密鑰/值對，其被收集作為減速機(jī)的輸出。缺省情況下，一行到第一個制表符的前綴是鍵，（不包括制表符）的其余部分的值為行。然而，這可以被定制為每次具體要求。

重要的命令