MapReduce 教學

Mapper

Mapper 將輸入鍵/值對應到一組中間鍵/值對。

Map 是將輸入記錄轉換為中間記錄的個別作業。轉換後的中間記錄不需要與輸入記錄為相同類型。給定的輸入對應到零或多個輸出對。

Hadoop MapReduce 框架會為作業的 InputFormat 所產生的每個 InputSplit 產生一個 Map 作業。

整體而言，會透過 Job.setMapperClass(Class) 方法將 Mapper 實作傳遞給作業。然後，框架會針對該作業的 InputSplit 中的每個鍵/值對呼叫 map(WritableComparable, Writable, Context)。然後，應用程式可以覆寫 cleanup(Context) 方法，以執行任何必要的清理工作。

輸出對不需要與輸入對為相同類型。給定的輸入對應到零或多個輸出對。輸出對會透過呼叫 context.write(WritableComparable, Writable) 來收集。

應用程式可以使用 Counter 來報告其統計資料。

與給定輸出鍵相關的所有中間值，隨後會由框架分組，並傳遞給 Reducer，以決定最終輸出。使用者可以透過 Job.setGroupingComparatorClass(Class) 指定 Comparator 來控制分組。

Mapper 輸出會排序，然後依 Reducer 分割。分割的總數與作業的縮減作業數相同。使用者可以透過實作自訂 Partitioner 來控制哪些鍵（以及記錄）會傳遞到哪些 Reducer。

使用者可以選擇透過 Job.setCombinerClass(Class) 指定 combiner，以執行中間輸出的區域聚合，有助於減少從 Mapper 傳輸到 Reducer 的資料量。

中間的排序輸出始終會儲存在簡單的（鍵長度、鍵、值長度、值）格式中。應用程式可以透過 Configuration 控制中間輸出是否要壓縮，以及如何壓縮，以及要使用的 CompressionCodec。

Reducer

Reducer 將一組共用一個金鑰的中間值縮減為一組較小的值。

使用者透過 Job.setNumReduceTasks(int) 設定工作中的縮減數量。

整體而言，Reducer 實作會透過 Job.setReducerClass(Class) 方法傳遞工作中的 Job，並可覆寫它來初始化自己。接著，架構會針對群組輸入中的每個 <金鑰，（值清單）> 對呼叫 reduce(WritableComparable, Iterable<Writable>, Context) 方法。接著，應用程式可以覆寫 cleanup(Context) 方法來執行任何必要的清理。

Reducer 有 3 個主要階段：洗牌、排序和縮減。

分割器

Partitioner 會分割金鑰空間。

Partitioner 控制中間映射輸出的金鑰分割。金鑰 (或金鑰的子集) 用於衍生分割，通常使用雜湊函數。分割的總數與作業的縮減任務數量相同。因此，這會控制中間金鑰 (以及記錄) 會傳送到哪一個 m 縮減任務中以進行縮減。

HashPartitioner 是預設的 Partitioner。

計數器

Counter 是 MapReduce 應用程式用來報告其統計資料的工具。

Mapper 和 Reducer 實作可以使用 Counter 來報告統計資料。

Hadoop MapReduce 內建一個 函式庫，其中包含一般有用的映射器、縮減器和分割器。

記憶體管理

使用者/管理員也可以使用 mapreduce.{map|reduce}.memory.mb 指定啟動子工作的最大虛擬記憶體，以及它遞迴啟動的任何子處理序。請注意，這裡設定的值是每個處理序的限制。mapreduce.{map|reduce}.memory.mb 的值應以 MB（百萬位元組）為單位指定。而且值必須大於或等於傳遞給 JavaVM 的 -Xmx，否則 VM 可能無法啟動。

注意：mapreduce.{map|reduce}.java.opts 僅用於配置 MRAppMaster 中啟動的子工作。在 配置 Hadoop Daemon 環境 中記載了配置 Daemon 記憶體選項的文件。

架構中某些部分可用的記憶體也是可配置的。在 MapReduce 工作中，調整影響作業並行性和資料存取磁碟頻率的參數可能會影響效能。監控工作的文件系統計數器，特別是相對於 MapReduce 中的位元組計數，對於調整這些參數非常有幫助。

Map 參數

從 Map 中發出的記錄會序列化成一個緩衝區，而元資料會儲存在會計緩衝區中。如以下選項所述，當序列化緩衝區或元資料超過臨界值時，緩衝區的內容會在背景中排序並寫入磁碟，而 Map 會繼續輸出記錄。如果在溢寫進行時任一緩衝區完全填滿，Map 執行緒會封鎖。當 Map 完成時，任何剩餘的記錄會寫入磁碟，而所有磁碟區段會合併成一個單一檔案。將溢寫到磁碟的次數降至最低可以減少 Map 時間，但較大的緩衝區也會減少 Map 程序可用的記憶體。

其他注意事項

• 如果在溢寫進行時超過任一溢寫臨界值，收集會持續到溢寫完成。例如，如果將 mapreduce.map.sort.spill.percent 設定為 0.33，而緩衝區的剩餘部分在溢寫執行時填滿，下一次溢寫會包含所有收集的記錄，或緩衝區的 0.66，而且不會產生其他溢寫。換句話說，這些臨界值定義的是觸發器，而不是封鎖。

• 大於序列化緩衝區的記錄會先觸發溢寫，然後溢寫到一個獨立的檔案。此記錄是否會先通過組合器是未定義的。

Shuffle/Reduce 參數

如前所述，每個縮減都會透過 HTTP 從區隔器取得指派給它的輸出並儲存在記憶體中，並定期將這些輸出合併到磁碟中。如果已開啟中間壓縮地圖輸出，每個輸出都會解壓縮到記憶體中。下列選項會影響縮減前這些合併到磁碟的頻率，以及縮減期間分配給地圖輸出的記憶體。

其他注意事項

• 如果映射輸出大於分配給複製映射輸出的記憶體的 25%，它將直接寫入磁碟，而不會先暫存在記憶體中。

• 當使用合併器執行時，關於高合併閾值和大型緩衝區的推理可能不成立。對於在擷取所有映射輸出之前啟動的合併，合併器會在溢出到磁碟時執行。在某些情況下，可以透過花費資源合併映射輸出（使磁碟溢出變小，並將溢出和擷取平行化）來獲得更好的縮減時間，而不是積極增加緩衝區大小。

• 當將記憶體中的映射輸出合併到磁碟以開始縮減時，如果需要中間合併，因為有要溢出的區段，並且磁碟上至少有 mapreduce.task.io.sort.factor 個區段，則記憶體中的映射輸出將會是中間合併的一部分。

已設定參數

下列屬性會在每個工作執行的工作設定中進行區域化

注意：在串流工作的執行期間，「mapreduce」參數的名稱會被轉換。點 ( . ) 會變成底線 ( _ )。例如，mapreduce.job.id 會變成 mapreduce_job_id，而 mapreduce.job.jar 會變成 mapreduce_job_jar。若要取得串流工作中的對應器/縮減器的值，請使用帶有底線的參數名稱。

工作記錄

NodeManager 會讀取工作的標準輸出 (stdout) 和錯誤 (stderr) 串流以及 syslog，並記錄到 ${HADOOP_LOG_DIR}/userlogs。

散布程式庫

「DistributedCache」也可以用來分發 jar 檔和原生程式庫，以便在 map 和/或 reduce 任務中使用。子 jvm 始終會將其「目前工作目錄」新增至 java.library.path 和 LD_LIBRARY_PATH。因此，快取的程式庫可透過 System.loadLibrary 或 System.load 載入。有關如何透過分散式快取載入共用程式庫的更多詳細資訊，請參閱 原生程式庫 中的說明。

工作控制

使用者可能需要串連 MapReduce 工作，以完成無法透過單一 MapReduce 工作執行的複雜任務。這相當容易，因為工作的輸出通常會傳送到分散式檔案系統，而輸出反過來可以當作下一個工作的輸入。

然而，這也表示確保工作完成（成功/失敗）的責任完全落在客戶端身上。在這種情況下，各種工作控制選項為

• Job.submit()：將工作提交到叢集並立即傳回。

• Job.waitForCompletion(boolean)：將工作提交到叢集並等待它完成。

InputSplit

InputSplit代表個別 Mapper 要處理的資料。

通常，InputSplit 呈現輸入的位元組導向檢視，而 RecordReader 則負責處理並呈現記錄導向檢視。

FileSplit 是預設的 InputSplit。它將 mapreduce.map.input.file 設定為邏輯分割的輸入檔案路徑。

RecordReader

RecordReader 從 InputSplit 讀取 <key, value> 配對。

通常，RecordReader 會轉換 InputSplit 提供的輸入位元組導向檢視，並為 Mapper 實作呈現記錄導向檢視以進行處理。因此，RecordReader 負責處理記錄邊界，並向工作提供金鑰和值。

OutputCommitter

OutputCommitter 描述 MapReduce 工作的任務輸出提交。

MapReduce 架構依賴工作的 OutputCommitter 來

1. 在初始化期間設定工作。例如，在工作的初始化期間建立工作的暫時輸出目錄。工作設定是由一個獨立的任務在工作處於 PREP 狀態且初始化任務之後完成。設定任務完成後，工作將移至 RUNNING 狀態。

2. 在工作完成後清理工作。例如，在工作完成後移除暫時輸出目錄。工作清理是由一個獨立的任務在工作結束時完成。清理任務完成後，工作會宣告 SUCCEDED/FAILED/KILLED。

3. 設定任務暫時輸出。任務設定是作為任務初始化期間的同一任務的一部分完成。

4. 檢查任務是否需要提交。這是為了在任務不需要提交時避免提交程序。

5. 提交任務輸出。任務完成後，任務會在需要時提交其輸出。

6. 捨棄工作提交。如果工作已失敗/終止，輸出將會清除。如果工作無法清除（在例外區塊中），將會啟動一個具有相同嘗試 ID 的獨立工作來執行清除。

FileOutputCommitter 是預設的 OutputCommitter。工作設定/清除工作會佔用映射或縮減容器，視 NodeManager 上哪一種可用而定。而且，JobCleanup 工作、TaskCleanup 工作和 JobSetup 工作具有最高優先順序，而且依此順序執行。

工作副作用檔案

在某些應用程式中，組件工作需要建立和/或寫入副作用檔案，這些檔案與實際工作輸出檔案不同。

在這種情況下，可能會出現兩個相同 Mapper 或 Reducer 執行個體同時執行（例如，推測性工作）的問題，它們會嘗試在 FileSystem 上開啟和/或寫入相同的檔案（路徑）。因此，應用程式撰寫者必須為每個工作嘗試（使用嘗試 ID，例如 attempt_200709221812_0001_m_000000_0）選擇唯一名稱，而不能只為每個工作選擇名稱。

為了避免這些問題，當 OutputCommitter 是 FileOutputCommitter 時，MapReduce 架構會在 FileSystem 上為每個工作嘗試維護一個特殊的 ${mapreduce.output.fileoutputformat.outputdir}/_temporary/_${taskid} 子目錄，可透過 ${mapreduce.task.output.dir} 存取，工作嘗試的輸出會儲存在此目錄中。在工作嘗試成功完成後，${mapreduce.output.fileoutputformat.outputdir}/_temporary/_${taskid}（僅此目錄）中的檔案會提升至 ${mapreduce.output.fileoutputformat.outputdir}。當然，架構會捨棄未成功工作嘗試的子目錄。這個程序對應用程式來說完全透明。

應用程式撰寫者可以在工作執行期間透過 FileOutputFormat.getWorkOutputPath(Conext) 在 ${mapreduce.task.output.dir} 中建立任何必要的副作用檔案來利用此功能，而架構會以類似的方式為成功的任務嘗試提升這些檔案，因此無需為每個任務嘗試選擇唯一的路徑。

請注意：在特定工作嘗試執行期間，${mapreduce.task.output.dir} 的值實際上是 ${mapreduce.output.fileoutputformat.outputdir}/_temporary/_{$taskid}，而且這個值是由 MapReduce 架構設定的。因此，只要在 MapReduce 工作傳回的 FileOutputFormat.getWorkOutputPath(Conext) 路徑中建立任何副作用檔案即可利用此功能。

由於在這種情況下，映射的輸出會直接傳送至 HDFS，因此整個討論也適用於縮減器為 NONE（即 0 個縮減器）的工作映射。

記錄寫入器

RecordWriter 會將輸出 <key, value> 對寫入輸出檔案。

RecordWriter 實作會將工作輸出寫入 FileSystem。

將工作提交至佇列

使用者會將工作提交至佇列。佇列作為工作的集合，允許系統提供特定功能。例如，佇列會使用 ACL 來控制哪些使用者可以向其提交工作。預期佇列主要由 Hadoop 排程器使用。

Hadoop 已設定為使用稱為「預設」的單一強制佇列。佇列名稱會定義在 Hadoop 站台設定的 mapreduce.job.queuename 屬性中。某些工作排程器，例如 容量排程器，支援多個佇列。

工作會透過 mapreduce.job.queuename 屬性或 Configuration.set(MRJobConfig.QUEUE_NAME, String) API 定義需要提交至的佇列。設定佇列名稱是選用的。如果提交工作時未附帶佇列名稱，則會提交至「預設」佇列。

計數器

Counters 代表全域計數器，由 MapReduce 架構或應用程式定義。每個 Counter 可以是任何 Enum 類型。特定 Enum 的計數器會分組為 Counters.Group 類型的群組。

應用程式可以定義任意 Counters (Enum 類型)，並透過 Counters.incrCounter(Enum, long) 或 Counters.incrCounter(String, String, long) 在 map 和/或 reduce 方法中更新它們。這些計數器會由架構進行全域彙總。

分散式快取

DistributedCache 有效地分發應用程式特定的、大型、唯讀檔案。

DistributedCache 是 MapReduce 架構提供的設施，用於快取應用程式需要的檔案 (文字、檔案、jar 等)。

應用程式會透過 Job 中的網址 (hdfs://) 指定要快取的檔案。DistributedCache 假設透過 hdfs:// 網址指定的檔案已存在於 FileSystem 中。

在工作中的任何工作在該節點上執行之前，架構會將必要的檔案複製到工作節點。其效率源自於檔案只會針對每個工作複製一次，以及解壓縮在工作節點上解壓縮的檔案的能力。

DistributedCache 會追蹤快取檔案的修改時間戳記。顯然在工作執行時，快取檔案不應由應用程式或外部修改。

DistributedCache 可用於分發簡單的唯讀資料/文字檔，以及更複雜的類型，例如檔案和 jar。檔案（zip、tar、tgz 和 tar.gz 檔）會在工作節點上「解壓縮」。檔案已設定「執行權限」。

檔案/檔案可透過設定屬性 mapreduce.job.cache.{files |archives} 來分發。如果必須分發多個檔案/檔案，它們可以作為逗號分隔的路徑新增。這些屬性也可以透過 API Job.addCacheFile(URI)/ Job.addCacheArchive(URI) 和 Job.setCacheFiles(URI[])/ Job.setCacheArchives(URI[]) 來設定，其中 URI 的格式為 hdfs://host:port/absolute-path#link-name。在串流中，檔案可透過命令列選項 -cacheFile/-cacheArchive 來分發。

DistributedCache 也可用作地圖和/或縮減任務中使用的基本軟體分發機制。它可用於分發 jar 和原生程式庫。Job.addArchiveToClassPath(Path) 或 Job.addFileToClassPath(Path) API 可用於快取檔案/jar，並將它們新增到子 jvm 的「classpath」中。透過設定組態屬性 mapreduce.job.classpath.{files |archives} 也能執行相同的動作。同樣地，連結到任務工作目錄中的快取檔案可用於分發原生程式庫並載入它們。

剖析

剖析是一種取得內建 Java 剖析器代表性範例 (2 或 3) 的公用程式，用於範例的對應和縮減。

使用者可以透過設定組態屬性 mapreduce.task.profile 來指定系統是否應收集工作中部分任務的剖析器資訊。可以使用 API Configuration.set(MRJobConfig.TASK_PROFILE, boolean) 設定值。如果值設定為 true，則會啟用任務剖析。剖析器資訊會儲存在使用者日誌目錄中。預設情況下，工作不會啟用剖析。

使用者在設定需要剖析後，可以使用組態屬性 mapreduce.task.profile.{maps|reduces} 來設定要剖析的 MapReduce 任務範圍。可以使用 API Configuration.set(MRJobConfig.NUM_{MAP|REDUCE}_PROFILES, String) 設定值。預設情況下，指定的範圍為 0-2。

使用者也可以透過設定組態屬性 mapreduce.task.profile.params 來指定剖析器組態引數。可以使用 API Configuration.set(MRJobConfig.TASK_PROFILE_PARAMS, String) 指定值。如果字串包含 %s，則會在任務執行時以剖析輸出檔案的名稱取代它。這些參數會透過命令列傳遞給任務子 JVM。剖析參數的預設值為 -agentlib:hprof=cpu=samples,heap=sites,force=n,thread=y,verbose=n,file=%s。

偵錯

MapReduce 架構提供一個讓使用者執行自訂指令碼以進行偵錯的功能。當 MapReduce 任務失敗時，使用者可以執行偵錯指令碼，例如處理任務日誌。指令碼會取得任務的 stdout 和 stderr 輸出、syslog 和 jobconf 的存取權。偵錯指令碼的 stdout 和 stderr 輸出會顯示在主控台診斷中，也會顯示在工作使用者介面的部分內容中。

在以下各節中，我們會說明如何提交偵錯指令碼和工作。指令碼檔案需要散布並提交至架構。

資料壓縮

Hadoop MapReduce 提供了讓應用程式撰寫者為中間 map 輸出和作業輸出（即 reduce 的輸出）指定壓縮的工具。它也隨附 CompressionCodec 實作，用於 zlib 壓縮演算法。gzip、bzip2、snappy 和 lz4 檔案格式也受支援。

Hadoop 也提供上述壓縮編解碼器的原生實作，原因在於效能（zlib）和 Java 函式庫不可用。有關其使用和可用性的更多詳細資訊，請參閱 這裡。

略過不良記錄

Hadoop 提供一個選項，在處理映射輸入時，可以略過一組不良輸入記錄。應用程式可透過 SkipBadRecords 類別控制此功能。

當映射工作在特定輸入上確定性地崩潰時，可以使用此功能。這通常是因映射函數中的錯誤所致。通常，使用者必須修正這些錯誤。然而，有時這並不可行。例如，錯誤可能出在第三方函式庫中，而其原始碼不可用。在這種情況下，即使經過多次嘗試，工作也永遠無法順利完成，且工作會失敗。有了這個功能，只有不良記錄周圍的一小部分資料會遺失，而這對某些應用程式來說可能是可以接受的（例如，對非常大型資料執行統計分析的應用程式）。

預設情況下，此功能已停用。若要啟用它，請參閱 SkipBadRecords.setMapperMaxSkipRecords(Configuration, long) 和 SkipBadRecords.setReducerMaxSkipGroups(Configuration, long)。

啟用此功能後，框架在一定數量的映射失敗後會進入「略過模式」。有關更多詳細資訊，請參閱 SkipBadRecords.setAttemptsToStartSkipping(Configuration, int)。在「略過模式」中，映射工作會維護正在處理的記錄範圍。為此，框架依賴於已處理記錄計數器。請參閱 SkipBadRecords.COUNTER_MAP_PROCESSED_RECORDS 和 SkipBadRecords.COUNTER_REDUCE_PROCESSED_GROUPS。此計數器使框架能夠知道已成功處理了多少記錄，因此，哪些記錄範圍導致工作崩潰。在後續嘗試中，會略過此記錄範圍。

略過的記錄數取決於應用程式增加已處理記錄計數器的頻率。建議在處理每筆記錄後增加此計數器。在通常會批次處理的某些應用程式中，這可能不可行。在這種情況下，框架可能會略過不良記錄周圍的其他記錄。使用者可以透過 SkipBadRecords.setMapperMaxSkipRecords(Configuration, long) 和 SkipBadRecords.setReducerMaxSkipGroups(Configuration, long) 控制略過記錄的數量。框架會嘗試使用類似二元搜尋的方法縮小略過記錄的範圍。略過的範圍會分成兩半，只執行其中一半。在後續失敗中，框架會找出哪一半包含不良記錄。工作會一直重新執行，直到達到可接受的略過值或耗盡所有工作嘗試。若要增加工作嘗試的數量，請使用 Job.setMaxMapAttempts(int) 和 Job.setMaxReduceAttempts(int)

略過的記錄會寫入 HDFS 中的順序檔案格式，以供後續分析。可以透過 SkipBadRecords.setSkipOutputPath(JobConf, Path) 變更位置。

原始碼

import java.io.BufferedReader;
import java.io.FileReader;
import java.io.IOException;
import java.net.URI;
import java.util.ArrayList;
import java.util.HashSet;
import java.util.List;
import java.util.Set;
import java.util.StringTokenizer;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.mapreduce.Counter;
import org.apache.hadoop.util.GenericOptionsParser;
import org.apache.hadoop.util.StringUtils;

public class WordCount2 {

  public static class TokenizerMapper
       extends Mapper<Object, Text, Text, IntWritable>{

    static enum CountersEnum { INPUT_WORDS }

    private final static IntWritable one = new IntWritable(1);
    private Text word = new Text();

    private boolean caseSensitive;
    private Set<String> patternsToSkip = new HashSet<String>();

    private Configuration conf;
    private BufferedReader fis;

    @Override
    public void setup(Context context) throws IOException,
        InterruptedException {
      conf = context.getConfiguration();
      caseSensitive = conf.getBoolean("wordcount.case.sensitive", true);
      if (conf.getBoolean("wordcount.skip.patterns", false)) {
        URI[] patternsURIs = Job.getInstance(conf).getCacheFiles();
        for (URI patternsURI : patternsURIs) {
          Path patternsPath = new Path(patternsURI.getPath());
          String patternsFileName = patternsPath.getName().toString();
          parseSkipFile(patternsFileName);
        }
      }
    }

    private void parseSkipFile(String fileName) {
      try {
        fis = new BufferedReader(new FileReader(fileName));
        String pattern = null;
        while ((pattern = fis.readLine()) != null) {
          patternsToSkip.add(pattern);
        }
      } catch (IOException ioe) {
        System.err.println("Caught exception while parsing the cached file '"
            + StringUtils.stringifyException(ioe));
      }
    }

    @Override
    public void map(Object key, Text value, Context context
                    ) throws IOException, InterruptedException {
      String line = (caseSensitive) ?
          value.toString() : value.toString().toLowerCase();
      for (String pattern : patternsToSkip) {
        line = line.replaceAll(pattern, "");
      }
      StringTokenizer itr = new StringTokenizer(line);
      while (itr.hasMoreTokens()) {
        word.set(itr.nextToken());
        context.write(word, one);
        Counter counter = context.getCounter(CountersEnum.class.getName(),
            CountersEnum.INPUT_WORDS.toString());
        counter.increment(1);
      }
    }
  }

  public static class IntSumReducer
       extends Reducer<Text,IntWritable,Text,IntWritable> {
    private IntWritable result = new IntWritable();

    public void reduce(Text key, Iterable<IntWritable> values,
                       Context context
                       ) throws IOException, InterruptedException {
      int sum = 0;
      for (IntWritable val : values) {
        sum += val.get();
      }
      result.set(sum);
      context.write(key, result);
    }
  }

  public static void main(String[] args) throws Exception {
    Configuration conf = new Configuration();
    GenericOptionsParser optionParser = new GenericOptionsParser(conf, args);
    String[] remainingArgs = optionParser.getRemainingArgs();
    if ((remainingArgs.length != 2) && (remainingArgs.length != 4)) {
      System.err.println("Usage: wordcount <in> <out> [-skip skipPatternFile]");
      System.exit(2);
    }
    Job job = Job.getInstance(conf, "word count");
    job.setJarByClass(WordCount2.class);
    job.setMapperClass(TokenizerMapper.class);
    job.setCombinerClass(IntSumReducer.class);
    job.setReducerClass(IntSumReducer.class);
    job.setOutputKeyClass(Text.class);
    job.setOutputValueClass(IntWritable.class);

    List<String> otherArgs = new ArrayList<String>();
    for (int i=0; i < remainingArgs.length; ++i) {
      if ("-skip".equals(remainingArgs[i])) {
        job.addCacheFile(new Path(remainingArgs[++i]).toUri());
        job.getConfiguration().setBoolean("wordcount.skip.patterns", true);
      } else {
        otherArgs.add(remainingArgs[i]);
      }
    }
    FileInputFormat.addInputPath(job, new Path(otherArgs.get(0)));
    FileOutputFormat.setOutputPath(job, new Path(otherArgs.get(1)));

    System.exit(job.waitForCompletion(true) ? 0 : 1);
  }
}

範例執行

範例文字檔案作為輸入

$ bin/hadoop fs -ls /user/joe/wordcount/input/
/user/joe/wordcount/input/file01
/user/joe/wordcount/input/file02

$ bin/hadoop fs -cat /user/joe/wordcount/input/file01
Hello World, Bye World!

$ bin/hadoop fs -cat /user/joe/wordcount/input/file02
Hello Hadoop, Goodbye to hadoop.

執行應用程式

$ bin/hadoop jar wc.jar WordCount2 /user/joe/wordcount/input /user/joe/wordcount/output

輸出

$ bin/hadoop fs -cat /user/joe/wordcount/output/part-r-00000
Bye 1
Goodbye 1
Hadoop, 1
Hello 2
World! 1
World, 1
hadoop. 1
to 1

請注意，輸入與我們查看的第一個版本不同，以及它們如何影響輸出。

現在，讓我們透過 DistributedCache 插入一個列出要忽略的字詞模式的模式檔案。

$ bin/hadoop fs -cat /user/joe/wordcount/patterns.txt
\.
\,
\!
to

再次執行它，這次使用更多選項

$ bin/hadoop jar wc.jar WordCount2 -Dwordcount.case.sensitive=true /user/joe/wordcount/input /user/joe/wordcount/output -skip /user/joe/wordcount/patterns.txt

正如預期，輸出

$ bin/hadoop fs -cat /user/joe/wordcount/output/part-r-00000
Bye 1
Goodbye 1
Hadoop 1
Hello 2
World 2
hadoop 1

再次執行它，這次關閉大小寫敏感性

$ bin/hadoop jar wc.jar WordCount2 -Dwordcount.case.sensitive=false /user/joe/wordcount/input /user/joe/wordcount/output -skip /user/joe/wordcount/patterns.txt

果然，輸出

$ bin/hadoop fs -cat /user/joe/wordcount/output/part-r-00000
bye 1
goodbye 1
hadoop 2
hello 2
world 2

重點

第二版的 WordCount 透過使用 MapReduce 架構提供的一些功能，改進了前一版

• 展示應用程式如何存取 Mapper (和 Reducer) 實作的 setup 方法中的組態參數。

• 展示如何使用 DistributedCache 來散發作業所需的唯讀資料。在此，它允許使用者指定在計算時要略過的字詞模式。

• 展示 GenericOptionsParser 的實用性，以處理一般 Hadoop 命令列選項。

• 展示應用程式如何使用 Counters，以及如何設定傳遞給 map (和 reduce) 方法的應用程式特定狀態資訊。

Java 和 JNI 是 Oracle America, Inc. 在美國和其他國家的商標或註冊商標。