前言

在這篇博客中，我們將學(xué)習(xí)如何使用 Java 實現(xiàn) Hadoop 的 MapReduce 框架，并通過一個詞頻統(tǒng)計（WordCount）的例子，來了解 MapReduce 的工作原理。MapReduce 是一種流行的大規(guī)模數(shù)據(jù)處理模式，廣泛應(yīng)用于分布式計算環(huán)境中。

一、正文

1. 代碼結(jié)構(gòu)

我們將在以下三個文件中實現(xiàn) MapReduce 的核心功能：

• Map.java: 實現(xiàn) Mapper 類，負責(zé)將輸入的文本數(shù)據(jù)按單詞進行拆分。
• Reduce.java: 實現(xiàn) Reducer 類，負責(zé)對單詞的出現(xiàn)次數(shù)進行匯總。
• WordCount.java: 設(shè)置作業(yè)（Job）配置，管理 Map 和 Reduce 的運行。

接下來我們將逐一分析這些代碼。

2. Map.java——Mapper 實現(xiàn)

首先看下 Mapper 類的代碼實現(xiàn)：

package demo1;

import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;

import java.io.IOException;
import java.util.StringTokenizer;


//public class Mapper<KEYIN, VALUEIN, KEYOUT, VALUEOUT>

public class Map extends Mapper<LongWritable, Text, Text, IntWritable> {
    private final static IntWritable one = new IntWritable(1); // 計數(shù)器
    private Text word = new Text(); // 存儲當(dāng)前處理的單詞

    @Override
    public void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
        // 將每行的文本數(shù)據(jù)分割成單詞,可使用split()實現(xiàn)相同功能
        StringTokenizer tokenizer = new StringTokenizer(value.toString());
        while (tokenizer.hasMoreTokens()) {
            word.set(tokenizer.nextToken()); // 獲取下一個單詞
            context.write(word, one); // 輸出單詞及其計數(shù)1
        }
    }
}

功能解讀：

• Mapper 的作用：
  Mapper 類的任務(wù)是將輸入的數(shù)據(jù)按行讀取，并對每一行的內(nèi)容進行處理。對于這個例子來說，我們的任務(wù)是將一行文本拆分成單詞，并為每個單詞標記它的初始計數(shù)值為 1。

• 重要方法與變量：

  • LongWritable key：表示輸入數(shù)據(jù)的偏移量，即每行文本在文件中的位置。
  • Text value：表示讀取的一行文本。
  • context.write(word, one)：將拆分出的單詞作為鍵（Text），值為 1（IntWritable），輸出到框架中供下一階段使用。

注意事項：

   StringTokenizer 用于分割每行文本，將其分割成單詞。

   context.write(word, one) 將結(jié)果輸出到 Reducer 處理時會被聚合。每遇到一個相同的單詞，后面會將其所有的 1 聚合成總和。

Mapper類的泛型定義

典型的 Mapper 類定義如下：

public class Mapper<KEYIN, VALUEIN, KEYOUT, VALUEOUT>

這表示 Mapper 是一個泛型類，帶有四個類型參數(shù)。每個參數(shù)對應(yīng) Mapper 任務(wù)中的不同數(shù)據(jù)類型。讓我們逐個解釋這些泛型參數(shù)的含義：

• KEYIN (輸入鍵的類型)：

  • 這是輸入數(shù)據(jù)的鍵的類型。在 MapReduce 程序中，輸入數(shù)據(jù)通常來自文件或其他形式的數(shù)據(jù)源，KEYIN 是表示該輸入數(shù)據(jù)片段的鍵。
  • 通常是文件中的偏移量（如文件的字節(jié)位置），所以經(jīng)常使用 Hadoop 提供的 LongWritable 來表示這個偏移量。

  常見類型：LongWritable，表示輸入文件中的行號或偏移量。

• VALUEIN (輸入值的類型)：

  • 這是輸入數(shù)據(jù)的值的類型。VALUEIN 是傳遞給 Mapper 的實際數(shù)據(jù)，通常是一行文本。
  • 通常是文件的內(nèi)容，比如一行文本，所以常用 Text 來表示。

  常見類型：Text，表示輸入文件中的一行文本。

• KEYOUT (輸出鍵的類型)：

  • 這是 Mapper 處理后的輸出數(shù)據(jù)的鍵的類型。Mapper 的輸出通常是某種鍵值對，KEYOUT 表示輸出鍵的類型。
  • 比如，在單詞計數(shù)程序中，輸出的鍵通常是一個單詞，所以常用 Text。

  常見類型：Text，表示處理后的單詞（在單詞計數(shù)程序中）。

• VALUEOUT (輸出值的類型)：

  • 這是 Mapper 處理后的輸出值的類型。VALUEOUT 是 Mapper 輸出鍵對應(yīng)的值的類型。
  • 在單詞計數(shù)程序中，輸出的值通常是一個數(shù)字，用于表示單詞的出現(xiàn)次數(shù)，所以常用 IntWritable。

  常見類型：IntWritable，表示單詞計數(shù)時的次數(shù)（1）。

3. Reduce.java——Reducer 實現(xiàn)

接下來我們實現(xiàn) Reducer：

package demo1;

import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.io.Text;

import java.io.IOException;

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

    @Override
    public void reduce(Text key, Iterable<IntWritable> values, Context context) throws IOException, InterruptedException {
        int sum = 0;
        for (IntWritable val : values) {
            sum += val.get(); // 累加單詞出現(xiàn)的次數(shù)
        }
        result.set(sum); // 設(shè)置聚合后的結(jié)果
        context.write(key, result); // 輸出單詞及其總次數(shù)
    }
}

功能解讀：

• Reducer 的作用：
  Reducer 類用于將 Mapper 輸出的單詞和它們的計數(shù)進行匯總。它會聚合每個單詞的所有 1，得到該單詞在整個輸入中的總計數(shù)。

• 重要方法與變量：

  • Text key：表示單詞。
  • Iterable<IntWritable> values：表示所有與該單詞相關(guān)聯(lián)的計數(shù)（1的集合）。
  • sum：用于累加該單詞出現(xiàn)的次數(shù)。
  • context.write(key, result)：輸出單詞及其出現(xiàn)的總次數(shù)。

注意事項：

  for (IntWritable val : values) 遍歷所有的計數(shù)值，并累加得到單詞的總次數(shù)。

      結(jié)果會輸出為 <單詞, 出現(xiàn)次數(shù)>，存儲到最終的輸出文件中。

Reducer類的泛型定義

public class Reducer<KEYIN, VALUEIN, KEYOUT, VALUEOUT>

Reducer 類帶有四個泛型參數(shù)，每個參數(shù)對應(yīng) Reducer 任務(wù)中的不同數(shù)據(jù)類型。

• KEYIN (輸入鍵的類型)：

  • KEYIN 是 Reducer 接收的鍵的類型，它是由 Mapper 的輸出鍵傳遞過來的類型。
  • 例如，在單詞計數(shù)程序中，Mapper 輸出的鍵是一個單詞，所以 KEYIN 通常是 Text 類型。

  常見類型：Text，表示單詞（在單詞計數(shù)程序中）。

• VALUEIN (輸入值的類型)：

  • VALUEIN 是 Reducer 接收的值的類型，它是 Mapper 輸出值的類型的集合。對于每個 KEYIN，Reducer 會接收一個與該鍵相關(guān)的值列表。
  • 例如，在單詞計數(shù)程序中，Mapper 輸出的值是每個單詞出現(xiàn)的次數(shù)（通常是 IntWritable 值為 1），所以 VALUEIN 的類型通常是 IntWritable。

  常見類型：IntWritable，表示單詞出現(xiàn)的次數(shù)。

• KEYOUT (輸出鍵的類型)：

  • KEYOUT 是 Reducer 輸出的鍵的類型。
  • 在單詞計數(shù)程序中，Reducer 輸出的鍵還是單詞，所以 KEYOUT 通常也是 Text 類型。

  常見類型：Text，表示單詞。

• VALUEOUT (輸出值的類型)：

  • VALUEOUT 是 Reducer 輸出的值的類型。這個值是 Reducer 處理后的結(jié)果。
  • 在單詞計數(shù)程序中，Reducer 輸出的值是每個單詞出現(xiàn)的總次數(shù)，所以 VALUEOUT 通常是 IntWritable。

  常見類型：IntWritable，表示單詞的總次數(shù)。

4. WordCount.java——作業(yè)配置與執(zhí)行

最后，我們編寫主程序，用于配置和啟動 MapReduce 作業(yè)：

package demo1;

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.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.util.GenericOptionsParser;

import java.io.IOException;

public class WordCount {
    public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException {
        Configuration conf = new Configuration(); // 配置項
        Job job = Job.getInstance(conf, "word count"); // 創(chuàng)建一個新作業(yè)

        String[] otherArgs = new GenericOptionsParser(conf, args).getRemainingArgs();
        if (otherArgs.length != 2) {
            System.err.println("Usage: wordcount <in> <out>");
            System.exit(1); // 輸入輸出路徑檢查
        }

        job.setJarByClass(WordCount.class); // 設(shè)置主類
        job.setMapperClass(Map.class); // 設(shè)置Mapper類
        job.setReducerClass(Reduce.class); // 設(shè)置Reducer類

        // 設(shè)置Map和Reduce輸出的鍵值對類型
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(IntWritable.class);

        // 輸入輸出路徑
        FileInputFormat.addInputPath(job, new Path(otherArgs[0])); // 輸入路徑
        FileOutputFormat.setOutputPath(job, new Path(otherArgs[1])); // 輸出路徑

        // 提交作業(yè)，直到作業(yè)完成
        System.exit(job.waitForCompletion(true) ? 0 : 1);
    }
}

功能解讀：

• 配置與作業(yè)初始化：
  • Configuration conf = new Configuration()：創(chuàng)建 Hadoop 的配置對象，存儲作業(yè)的相關(guān)配置。
  • Job job = Job.getInstance(conf, "word count")：創(chuàng)建 MapReduce 作業(yè)，并為作業(yè)命名為 "word count"。
• 作業(yè)設(shè)置：
  • job.setJarByClass(WordCount.class)：設(shè)置運行時的主類。
  • job.setMapperClass(Map.class) 和 job.setReducerClass(Reduce.class)：分別設(shè)置 Mapper 和 Reducer 類。
• 輸入輸出路徑：
  • FileInputFormat.addInputPath()：指定輸入數(shù)據(jù)的路徑。
  • FileOutputFormat.setOutputPath()：指定輸出結(jié)果的路徑。
• 作業(yè)提交與運行：
  • System.exit(job.waitForCompletion(true) ? 0 : 1)：提交作業(yè)并等待完成，如果成功返回 0，失敗返回 1。

注意事項：

   GenericOptionsParser 用于解析命令行輸入，獲取輸入和輸出路徑。

         提交作業(yè)后，Hadoop 框架會根據(jù)配置自動運行 Mapper 和 Reducer，并將結(jié)果輸出到指定的路徑。

二、知識回顧與補充

數(shù)據(jù)的偏移量？

數(shù)據(jù)的偏移量，即 LongWritable key，是 MapReduce 程序中 Mapper 輸入的鍵，它表示輸入數(shù)據(jù)文件中每行文本的起始字節(jié)位置。

例：假設(shè)我們有一個文本文件：

0: Hello world
12: Hadoop MapReduce
32: Data processing

        每行的前面的數(shù)字（0, 12, 32）就是對應(yīng)行在文件中的偏移量，表示從文件開頭到該行起始字節(jié)的距離。LongWritable 類型的 key 就表示這個偏移量。

        在 Mapper 中，輸入是以 <偏移量, 文本行> 這樣的鍵值對形式提供的。雖然偏移量在詞頻統(tǒng)計任務(wù)中不重要，但在某些應(yīng)用中，如文件處理、日志解析時，偏移量可以幫助追蹤數(shù)據(jù)的位置。

Context

Context 是 MapReduce 框架中 Mapper 和 Reducer 中非常重要的一個類，它提供了與框架進行交互的方法。

Context 的主要作用：

• 寫入輸出：在 Mapper 和 Reducer 中，context.write(key, value) 用于將結(jié)果輸出到框架?？蚣軙詣犹幚磉@些輸出結(jié)果，并將 Mapper 的輸出作為 Reducer 的輸入，或者將最終的 Reducer 輸出保存到 HDFS。

  • 在 Mapper 中，context.write(word, one) 將每個單詞及其初始計數(shù) 1 傳遞給框架。
  • 在 Reducer 中，context.write(key, result) 將每個單詞及其總出現(xiàn)次數(shù)輸出到最終結(jié)果。

• 配置訪問：Context 可以訪問作業(yè)的配置參數(shù)（如 Configuration），幫助程序獲取環(huán)境變量或作業(yè)參數(shù)。

• 計數(shù)器：Context 提供計數(shù)器（counter）的支持，用于統(tǒng)計作業(yè)中的某些事件（如錯誤次數(shù)、特定條件的滿足次數(shù)等）。

• 記錄狀態(tài)：Context 可以報告作業(yè)的執(zhí)行狀態(tài)，幫助開發(fā)者追蹤作業(yè)的進度或調(diào)試

Iterable<IntWritable> values是什么類型？

在 Reducer 階段，Iterable<IntWritable> values 表示與同一個鍵（即單詞）相關(guān)聯(lián)的所有 IntWritable 值的集合。

• 類型解讀：

  • Iterable 表示一個可以迭代的集合，意味著它可以被遍歷。
  • IntWritable 是 Hadoop 定義的一個包裝類，用于封裝 int 類型的值。

• 在詞頻統(tǒng)計的例子中：
  對于每個單詞，Mapper 會輸出多個 <單詞, 1>，因此在 Reducer 中，對于每個鍵（即單詞），會有多個 1 作為值的集合，即 values。Reducer 的任務(wù)就是對這些 1 進行累加，計算單詞的總出現(xiàn)次數(shù)。

其他遍歷方法：

除了 for (IntWritable val : values) 這種增強型 for 循環(huán)，我們還可以使用 Iterable遍歷

Iterator<IntWritable> iterator = values.iterator();
while (iterator.hasNext()) {
    IntWritable val = iterator.next();
    sum += val.get(); // 處理每個值
}

Iterator 提供 hasNext() 方法檢查是否有更多元素，next() 方法返回當(dāng)前元素并指向下一個。

Configuration conf = new Configuration() 的作用是什么？

Configuration 類用于加載和存儲 Hadoop 應(yīng)用程序運行時的配置信息，它是一個 Hadoop 配置系統(tǒng)的核心組件，能夠讓你定義和訪問一些運行時參數(shù)。每個 MapReduce 作業(yè)都依賴 Configuration 來初始化作業(yè)配置。

Configuration 的具體作用：

• 讀取配置文件：

  • 它默認會加載系統(tǒng)的 Hadoop 配置文件，如 core-site.xml、hdfs-site.xml、mapred-site.xml 等，這些文件包含了 Hadoop 集群的信息（如 HDFS 的地址、作業(yè)調(diào)度器等）。
  • 如果需要，可以通過代碼手動添加或覆蓋這些參數(shù)。

• 自定義參數(shù)傳遞：

  • 你可以在運行 MapReduce 作業(yè)時通過 Configuration 傳遞一些自定義參數(shù)。例如，你可以將某些控制邏輯寫入配置文件或直接在代碼中設(shè)置特定參數(shù)，并在 Mapper 或 Reducer 中通過 context.getConfiguration() 來訪問這些參數(shù)。
  • 示例：

    Configuration conf = new Configuration();
    conf.set("my.custom.param", "some value");
    

• 作業(yè)設(shè)置的依賴：

  • Configuration 是 Hadoop 作業(yè)運行的基礎(chǔ)，它為 Job 提供上下文，包括輸入輸出格式、作業(yè)名稱、運行時依賴庫等等。

為什么需要 Configuration？

在 MapReduce 應(yīng)用中，集群的規(guī)模較大，許多配置參數(shù)（如文件系統(tǒng)路徑、任務(wù)調(diào)度器配置等）都存儲在外部的配置文件中，Configuration 類可以動態(tài)加載這些配置，避免硬編碼。

用 split() 方法實現(xiàn)默認分隔符的分割

如果想實現(xiàn)類似于 StringTokenizer 的默認行為（用空白字符分割），可以使用正則表達式 \\s+，它表示匹配一個或多個空白字符（與 StringTokenizer 的默認行為一樣）。

示例代碼：

    @Override
    public void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
        // 使用 split() 方法來分割字符串，使用空白字符作為分隔符
        String[] tokens = value.toString().split("\\s+");

        // 遍歷分割后的標記
        for (String token : tokens) {
            word.set(token);
            context.write(word, one);
        }
    }

IntWritable

IntWritable 是 Hadoop 提供的一個類，屬于 org.apache.hadoop.io 包。它是 Hadoop 框架中用來封裝 int 類型的數(shù)據(jù)的一個可序列化（writable）的包裝類。

在 Hadoop MapReduce 中，所有數(shù)據(jù)類型都需要實現(xiàn) Writable 和 Comparable 接口，以便能夠通過網(wǎng)絡(luò)在節(jié)點之間傳輸。IntWritable 作為 Hadoop 中的基本數(shù)據(jù)類型之一，提供了一些便利方法來存儲和處理 int 數(shù)據(jù)。

IntWritable 類的作用：

在 MapReduce 中，Hadoop的數(shù)據(jù)類型都需要實現(xiàn) Writable 接口，這樣它們就可以在分布式系統(tǒng)中通過網(wǎng)絡(luò)傳輸。IntWritable 封裝了一個 Java 的 int 類型，用于 Hadoop 的輸入輸出鍵值對。

主要的用途：

• MapReduce 中作為值類型：IntWritable 常用于表示 Mapper 和 Reducer 的輸出值。
• 支持序列化和反序列化：它實現(xiàn)了 Writable 接口，可以在分布式環(huán)境下高效地進行序列化和反序列化。

如何使用 IntWritable

IntWritable 提供了構(gòu)造方法和一些方法來設(shè)置和獲取 int 值。

1. 創(chuàng)建 IntWritable 對象

可以通過構(gòu)造方法直接創(chuàng)建對象：

• 默認構(gòu)造函數(shù)：創(chuàng)建一個值為 0 的 IntWritable。
• 參數(shù)構(gòu)造函數(shù)：可以直接設(shè)置初始值。

// 創(chuàng)建一個默認值為 0 的 IntWritable 對象
IntWritable writable1 = new IntWritable();

// 創(chuàng)建一個值為 10 的 IntWritable 對象
IntWritable writable2 = new IntWritable(10);

2. 設(shè)置值和獲取值

可以通過 set() 方法來設(shè)置值，通過 get() 方法來獲取 IntWritable 封裝的 int 值。

IntWritable writable = new IntWritable();

// 設(shè)置值為 42
writable.set(42);

// 獲取值
int value = writable.get(); // value == 42

3. 在 MapReduce 中使用

在 MapReduce 任務(wù)中，IntWritable 通常被用于輸出的值。例如，在計數(shù)器的 MapReduce 程序中，常將 IntWritable 的值設(shè)置為 1，表示一個單詞的出現(xiàn)次數(shù)。

示例：MapReduce 中的 IntWritable 使用

public class Map extends Mapper<LongWritable, Text, Text, IntWritable> {
    private final static IntWritable one = new IntWritable(1); // 值為 1 的 IntWritable
    private Text word = new Text();

    @Override
    public void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
        String[] tokens = value.toString().split("\\s+");

        // 遍歷每個單詞并寫入 context
        for (String token : tokens) {
            word.set(token);
            context.write(word, one);  // 輸出 <單詞, 1>
        }
    }
}

在這個 MapReduce 例子中：

• 每個單詞對應(yīng)的值都是 1，使用 IntWritable one = new IntWritable(1) 來封裝這個整數(shù)值。
• context.write() 將 Text 和 IntWritable 對象作為鍵值對輸出，鍵是單詞，值是 1。

總結(jié)：

IntWritable 之所以在 Hadoop 中使用，而不是原生的 int 類型，是因為：

        Hadoop 需要能通過網(wǎng)絡(luò)傳輸?shù)念愋停?code>IntWritable 實現(xiàn)了 Writable 接口，可以序列化和反序列化。

  IntWritable 實現(xiàn)了 Comparable 接口，因此可以在 Hadoop 的排序操作中使用。

Job job = Job.getInstance(conf, "word count"); 這是什么意思？

Job job = Job.getInstance(conf, "word count");

這行代碼創(chuàng)建并配置一個新的 MapReduce 作業(yè)實例。

• conf：這是一個 Hadoop 的 Configuration 對象，包含了作業(yè)的配置信息。
• "word count"：這是作業(yè)的名稱，可以是任何字符串。它主要用于標識和記錄作業(yè)。

在Driver中，為什么只設(shè)置輸出的鍵值對類型？不設(shè)置輸入呢？

job.setOutputKeyClass(Text.class);
job.setOutputValueClass(IntWritable.class);

        1. 輸入數(shù)據(jù)的鍵值對類型

        是由 InputFormat（如 TextInputFormat）決定的，默認讀取每行數(shù)據(jù)的偏移量和內(nèi)容作為鍵值對傳遞給 Mapper。 

        Hadoop MapReduce 使用 InputFormat 類來讀取輸入數(shù)據(jù)文件。默認的輸入格式是 TextInputFormat，它會自動將輸入文件解析成鍵值對形式，而你不需要在 Driver 中顯式指定輸入的類型。

• TextInputFormat 的輸出（即傳遞給 Mapper 的輸入）是：
  • 鍵：每一行文本在文件中的字節(jié)偏移量，類型為 LongWritable。
  • 值：每一行的內(nèi)容，類型為 Text。

所以，Mapper 的輸入鍵值對類型已經(jīng)由 InputFormat 控制，不需要你在 Driver 中手動指定。

        2. 最終輸出的鍵值對類型

        需要你在 Driver 中顯式設(shè)置，因為這是寫入到 HDFS 中的數(shù)據(jù)類型。

 setOutputKeyClass 和 setOutputValueClass 的作用

        在 Driver 中，你需要明確指定的是 最終輸出結(jié)果的鍵值對類型，即 Reducer 輸出的鍵值對類型，因為這是寫入到 HDFS 中的數(shù)據(jù)類型。

• job.setOutputKeyClass(Text.class)：指定 最終輸出的鍵 類型為 Text。
• job.setOutputValueClass(IntWritable.class)：指定 最終輸出的值 類型為 IntWritable。

這兩項設(shè)置明確告訴 Hadoop，最后存儲在 HDFS 中的結(jié)果文件中，鍵和值分別是什么類型。

總結(jié)

到此這篇關(guān)于使用Java實現(xiàn)MapReduce詞頻統(tǒng)計的文章就介紹到這了,更多相關(guān)Java MapReduce詞頻統(tǒng)計內(nèi)容請搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家！

最新評論