Hadoop數據流是Hadoop自帶發行的實用程序。該實用程序允許創建和運行Map/Reduce任務的任何可執行文件或腳本映射器和/或減速器。

使用Python示例

對於Hadoop的數據流，我們考慮的字計數問題。任何工作在Hadoop中必須有兩個階段：映射器和減速器。我們使用python腳本代碼映射器和減速器在Hadoop下運行它。使用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)

請確保此文件具有執行權限（使用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 的主目錄映射器和減速器代碼。確保這些文件具有執行權限（使用chmod +x mapper.py 和 chmod +x reducer.py）。由於python具有大小寫敏感，因此相同的代碼可以從以下鏈接下載。

wordCount程序的執行

$ $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

其中“\”用於續行以便於閱讀。

例如，

./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

數據流工作原理

在上麵的例子中，這兩個映射器和減速是從標準輸入讀取作為輸入，並輸出到標準輸出到Python腳本。實用程序將創建一個Map/Reduce作業，並將作業提交到一個合適的集群，並監督工作的進展情況，直至完成。

當指定映射器的腳本，每個映射任務將啟動腳本作為一個單獨的進程時映射器初始化。作為mapper任務運行時，輸入轉換成行給進程的標準輸入（STDIN）。在此期間，映射器收集從該方法的標準輸出（stdout）麵向行輸出和每一行轉換為鍵/值對，其被收集作為映射器的輸出。缺省情況下，一行到第一個製表符的前綴是鍵和行（不包括製表符）的其餘部分為值。如果在該行冇有任何製表符，則整行鍵和值被視為null。然而，這可以被定製，每次需要1個。

當指定減速腳本，每個減速器任務將啟動腳本作為一個單獨的進程，然後減速初始化。減速器任務運行時將其轉換其輸入鍵/值對，進入行並將該行進程的標準輸入（STDIN）。在此期間，在減速機收集來自該過程的標準輸出（stdout）的麵向行的輸出，每行轉換成一個密鑰/值對，其被收集作為減速機的輸出。缺省情況下，一行到第一個製表符的前綴是鍵，（不包括製表符）的其餘部分的值為行。然而，這可以被定製為每次具體要求。

重要的命令

參數	描述
-input directory/file-name	輸入位置映射。 （必填）
-output directory-name	輸出位置的減速器。 （必填）
-mapper executable or script or JavaClassName	映射器可執行文件。 （必填）
-reducer executable or script or JavaClassName	減速器的可執行文件。 （必填）
-file file-name	使現有的映射器，減速機，或組合的可執行本地計算節點上。
-inputformat JavaClassName	類，應該提供返回鍵/值對文字類。如果冇有指定，使用TextInputFormat作為默認。
-outputformat JavaClassName	類，提供應采取鍵/值對文字類的。如果冇有指定，使用TextOutputformat作為默認值。
-partitioner JavaClassName	類，確定哪個減少一個鍵被發送。
-combiner streamingCommand or JavaClassName	組合可執行文件映射輸出。
-cmdenv name=value	通過環境變量數據流的命令。
-inputreader	對於向後兼容性：指定記錄讀取器類（而不是輸入格式類）。
-verbose	詳細的輸出。
-lazyOutput	創建懶輸出。例如，如果輸出格式是基於FileOutputFormat，輸出文件僅在第一次調用output.collect（或Context.write）創建。
-numReduceTasks	指定減速器的數目。
-mapdebug	當map任務失敗的腳本調用。
-reducedebug	腳本調用時降低任務失敗。