MapReduce編程實踐

重要知識點：

1. MapReduce是一種分布式并行編程模型,是Hadoop核心子專案之一,如果已經安裝了Hadoop，就不需要另外安裝MapReduce，
2. 主要的理論知識點包括：MapReduce概述、MapReduce的作業流程，WordCount實體分析，MapReduce的具體應用，
3. 掌握MapReduce的核心思想、編程模型、作業原理和實體分析，
4. MapReduce的程式可以用Eclipse編譯運行或使用命令列編譯打包運行，本實驗使用Eclipse編譯運行MapReduce程式，

實驗內容與步驟：

一、Eclipse的安裝與配置
該部分實驗前面已經做過，省略
https://blog.csdn.net/weixin_43640161/article/details/108691921

二、Hadoop-Eclipse-Plugin的安裝與配置

1. 安裝 Hadoop-Eclipse-Plugin

要在 Eclipse 上編譯和運行 MapReduce 程式，需要安裝 hadoop-eclipse-plugin
下載地址：http://pan.baidu.com/s/1i4ikIoP

下載后，將 release 中的 hadoop-eclipse-kepler-plugin-2.6.0.jar （還提供了 2.2.0 和 2.4.1 版本）復制到 Eclipse 安裝目錄的 plugins 檔案夾中

終端命令：sudo mv hadoop-eclipse-plugin-2.6.0.jar /opt/eclipse/plugins/

運行 eclipse -clean 重啟 Eclipse 即可（添加插件后只需要運行一次該命令，以后按照正常方式啟動就行了），

終端命令：eclipse -clean

提示：如果對命令不是很熟悉，也可以手動找到jar包直接拷貝過去，然后找到安裝檔案目錄雙擊重新啟動eclipse完成第一步操作，

2. 配置 Hadoop-Eclipse-Plugin

在繼續配置前請確保已經開啟了 Hadoop，
終端命令： start-all.sh

啟動 Eclipse 后就可以在左側的Project Explorer中看到 DFS Locations（若看到的是 welcome 界面，點擊左上角的 x 關閉就可以看到了，

沒用顯示的話，可以查看


看到以上畫面也算成功

插件需要進一步的配置，

第一步：選擇 Window 選單下的 Preference，

此時會彈出一個表單，表單的左側會多出 Hadoop Map/Reduce 選項，點擊此選項，選擇 Hadoop 的安裝目錄（如/bigdata/hadoop，如果不好選擇目錄，直接輸入就行），

第二步：切換 Map/Reduce 開發視圖，選擇 Window 選單下選擇 Window -> Perspective -> Open Perspective -> Other，彈出一個表單，從中選擇 Map/Reduce 選項即可進行切換，

第三步：建立與 Hadoop 集群的連接，點擊 Eclipse軟體右下角的 Map/Reduce Locations 面板，在面板中單擊右鍵，選擇 New Hadoop Location，

在彈出來的 General 選項面板中，General 的設定要與 Hadoop 的配置一致，一般兩個 Host 值是一樣的，如果是偽分布式，填寫 localhost 即可，另外我使用的Hadoop偽分布式配置，設定 fs.defaultFS 為 hdfs://localhost:9000，則 DFS Master 的 Port 要改為 9000，Map/Reduce(V2) Master 的 Port 用默認的即可，Location Name 隨意填寫，

最后的設定如下圖所示：

Advanced parameters 選項面板是對 Hadoop 引數進行配置，實際上就是填寫 Hadoop 的配置項(/bigdata/hadoop3.1.1/etc/hadoop中的組態檔)，如我配置了 hadoop.tmp.dir ，就要進行相應的修改，但修改起來會比較繁瑣，我們可以通過復制組態檔的方式解決（下面會說到），

總之，我們只要配置 General 就行了，點擊 finish，Map/Reduce Location 就創建好了，

三、在 Eclipse 中操作 HDFS 中的檔案

配置好后，點擊左側 Project Explorer 中的 MapReduce Location （點擊三角形展開）就能直接查看 HDFS 中的檔案串列了（HDFS 中要有檔案，如下圖是 WordCount 的輸出結果），雙擊可以查看內容，右鍵點擊可以上傳、下載、洗掉 HDFS 中的檔案，無需再通過繁瑣的 hdfs dfs -ls 等命令進行操作了，
以下input/myLocalFile.txt檔案記錄了檔案結果，

如果無法查看，可右鍵點擊 Location 嘗試 Reconnect 或重啟 Eclipse，

Tips：
HDFS 中的內容變動后，Eclipse 不會同步重繪，需要右鍵點擊 Project Explorer中的 MapReduce Location，選擇 Refresh，才能看到變動后的檔案，

四、在 Eclipse 中創建 MapReduce 專案
點擊 File 選單，選擇 New -> Project…:選擇 Map/Reduce Project，點擊 Next，

填寫 Project name 為 WordCount 即可，點擊 Finish 就創建好了專案，

此時在左側的 Project Explorer 就能看到剛才建立的專案了，

接著右鍵點擊剛創建的 WordCount 專案src，選擇 New -> Class，需要填寫兩個地方：在 Package 處填寫 org.apache.hadoop.examples；在 Name 處填寫 WordCount，

創建 Class 完成后，在 Project 的 src 中就能看到 WordCount.java 這個檔案，將如下 WordCount 的代碼復制到該檔案中，

package org.apache.hadoop.examples;
 
import java.io.IOException;
import java.util.Iterator;
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.util.GenericOptionsParser;
 
public class WordCount {
    public WordCount() {
    }
 
    public static void main(String[] args) throws Exception {
        Configuration conf = new Configuration();
        String[] otherArgs = (new GenericOptionsParser(conf, args)).getRemainingArgs();
        if(otherArgs.length < 2) {
            System.err.println("Usage: wordcount <in> [<in>...] <out>");
            System.exit(2);
        }
 
        Job job = Job.getInstance(conf, "wordcount");
        job.setJarByClass(WordCount.class);
        job.setMapperClass(WordCount.TokenizerMapper.class);
        job.setCombinerClass(WordCount.IntSumReducer.class);
        job.setReducerClass(WordCount.IntSumReducer.class);
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(IntWritable.class);
 
        for(int i = 0; i < otherArgs.length - 1; ++i) {
            FileInputFormat.addInputPath(job, new Path(otherArgs[i]));
        }
 
        FileOutputFormat.setOutputPath(job, new Path(otherArgs[otherArgs.length - 1]));
        System.exit(job.waitForCompletion(true)?0:1);
    }
 
    public static class IntSumReducer extends Reducer<Text, IntWritable, Text, IntWritable> {
        private IntWritable result = new IntWritable();
 
        public IntSumReducer() {
        }
 
        public void reduce(Text key, Iterable<IntWritable> values, Reducer<Text, IntWritable, Text, IntWritable>.Context context) throws IOException, InterruptedException {
            int sum = 0;
 
            IntWritable val;
            for(Iterator i$ = values.iterator(); i$.hasNext(); sum += val.get()) {
                val = (IntWritable)i$.next();
            }
 
            this.result.set(sum);
            context.write(key, this.result);
        }
    }
 
    public static class TokenizerMapper extends Mapper<Object, Text, Text, IntWritable> {
        private static final IntWritable one = new IntWritable(1);
        private Text word = new Text();
 
        public TokenizerMapper() {
        }
 
        public void map(Object key, Text value, Mapper<Object, Text, Text, IntWritable>.Context context) throws IOException, InterruptedException {
            StringTokenizer itr = new StringTokenizer(value.toString());
 
            while(itr.hasMoreTokens()) {
                this.word.set(itr.nextToken());
                context.write(this.word, one);
            }
 
        }
    }
}

五、通過 Eclipse 運行 MapReduce

在運行 MapReduce 程式前，還需要執行一項重要操作（也就是上面提到的通過復制組態檔解決引數設定問題）：將 /bigadata/hadoop3.1.1/etc/hadoop 中將有修改過的組態檔（如偽分布式需要 core-site.xml 和 hdfs-site.xml），以及 log4j.properties 復制到 WordCount 專案下的 src 檔案夾（~/workspace/WordCount/src）中：

終端命令：
cp core-site.xml ~/workspace/WordCount/src
cp hdfs-site.xml ~/workspace/WordCount/src
cp log4j.properties ~/workspace/WordCount/src

提示：上述操作也可以手動找到相關檔案，直接拷貝粘貼過去，然后重繪即可，
沒有復制這些檔案的話程式將無法正確運行，本實驗最后再解釋為什么需要復制這些檔案，

復制完成后，務必右鍵點擊 WordCount 選擇 refresh 進行重繪（不會自動重繪，需要手動重繪），可以看到檔案結構如下所示：

點擊工具列中的 Run 圖示，或者右鍵點擊 Project Explorer 中的 WordCount.java，選擇 Run As -> Run on Hadoop，就可以運行 MapReduce 程式了，不過由于沒有指定引數，運行時會提示 “Usage: wordcount “，需要通過Eclipse設定一下運行引數，

右鍵點擊剛創建的 WordCount.java，選擇 Run As -> Run Configurations，在此處可以設定運行時的相關引數（如果 Java Application 下面沒有 WordCount，那么需要先雙擊 Java Application），切換到 “Arguments” 欄，在 Program arguments 處填寫 “input output” 就可以了，

或者也可以直接在代碼中設定好輸入引數，可將代碼 main() 函式的 String[] otherArgs = new GenericOptionsParser(conf, args).getRemainingArgs(); 改為：
// String[] otherArgs = new GenericOptionsParser(conf, args).getRemainingArgs();
String[] otherArgs=new String[]{“input”,“output”}; /* 直接設定輸入引數 */
至此，你就可以使用 Eclipse 方便的進行 MapReduce程式的開發了，

六、單詞統計測驗

新建一個檔案input,并寫入4行單詞，如下：
首先，使用vim編輯器，在本地Linux檔案系統的“/home/hadoop/”目錄下創建一個檔案input，里面可以隨意輸入一些單詞，比如，輸入如下四行：

hello hadoop
hello hbase
hello mapreduce
hello hdfs hbase

然后，可以使用如下命令把本地檔案系統的“/home/hadoop/input”上傳到HDFS中的當前用戶目錄的根目錄下，也就是上傳到HDFS的“/user/hadoop/”目錄下，然后查看是否上傳成功，

終端命令：
hdfs dfs -put ./input /user/hadoop（這里的hadoop是我的用戶名）
hdfs dfs -ls -R
hdfs dfs -text input


注意：如果之前實驗遺留的input檔案夾，會使上傳出錯，要提前洗掉，
終端命令：hdfs dfs -rm -r /user/hadoop/input
重新運行WordCount.java檔案，如果提示成功，重繪 DFS Location 后就能看到輸出的 output 檔案夾，（建議重啟Eclipse）

七、在 Eclipse 中運行 MapReduce 程式會遇到的問題

在使用 Eclipse 運行 MapReduce 程式時，會讀取 Hadoop-Eclipse-Plugin 的 Advanced parameters 作為 Hadoop 運行引數，如果我們未進行修改，則默認的引數其實就是單機（非分布式）引數，因此程式運行時是讀取本地目錄而不是 HDFS 目錄，就會提示 Input 路徑不存在，

Exception in thread “main” org.apache.hadoop.mapreduce.lib.input.InvalidInputException: Input path does not exist: file:/home/hadoop/workspace/WordCountProject/input
所以我們需要將組態檔復制到專案中的 src 目錄，來覆寫這些引數，讓程式能夠正確運行，

log4j 用于記錄程式的輸出日記，需要 log4j.properties 這個組態檔，如果沒有復制該檔案到專案中，運行程式后在 Console 面板中會出現警告提示：

log4j:WARN No appenders could be found for logger (org.apache.hadoop.metrics2.lib.MutableMetricsFactory).
log4j:WARN Please initialize the log4j system properly.
log4j:WARN See http://logging.apache.org/log4j/1.2/faq.html#noconfig for more info.
雖然不影響程式的正確運行的，但程式運行時無法看到任何提示訊息（只能看到出錯資訊），

到了這一步，本次實驗就完成了，你今天學會了嗎？