大數據【四】MapReduce（單詞計數；二次排序；計數器；join；分布式緩存）

   　前言：

　　　　根據前面的幾篇博客學習，現在可以進行MapReduce學習了。本篇博客首先闡述了MapReduce的概念及使用原理，其次直接從五個實驗中實踐學習（單詞計數，二次排序，計數器，join，分布式緩存）。

 

一 概述

定義

MapReduce是一種計算模型，簡單的說就是將大批量的工作（數據）分解（MAP）執行，然后再將結果合並成最終結果（REDUCE）。這樣做的好處是可以在任務被分解后，可以通過大量機器進行並行計算，減少整個操作的時間。

適用范圍：數據量大，但是數據種類小可以放入內存。

基本原理及要點：將數據交給不同的機器去處理，數據划分，結果歸約。

理解MapReduce和Yarn：在新版Hadoop中，Yarn作為一個資源管理調度框架，是Hadoop下MapReduce程序運行的生存環境。其實MapRuduce除了可以運行Yarn框架下，也可以運行在諸如Mesos，Corona之類的調度框架上，使用不同的調度框架，需要針對Hadoop做不同的適配。（了解YARN見上一篇博客>>　　http://www.cnblogs.com/1996swg/p/7286490.html　　）

MapReduce編程

編寫在Hadoop中依賴Yarn框架執行的MapReduce程序，並不需要自己開發MRAppMaster和YARNRunner，因為Hadoop已經默認提供通用的YARNRunner和MRAppMaster程序， 大部分情況下只需要編寫相應的Map處理和Reduce處理過程的業務程序即可。

編寫一個MapReduce程序並不復雜，關鍵點在於掌握分布式的編程思想和方法，主要將計算過程分為以下五個步驟：

（1）迭代。遍歷輸入數據，並將之解析成key/value對。

（2）將輸入key/value對映射(map)成另外一些key/value對。

（3）依據key對中間數據進行分組(grouping)。

（4）以組為單位對數據進行歸約(reduce)。

（5）迭代。將最終產生的key/value對保存到輸出文件中。

 Java API解析

（1）InputFormat：用於描述輸入數據的格式，常用的為TextInputFormat提供如下兩個功能：

數據切分： 按照某個策略將輸入數據切分成若干個split，以便確定Map Task個數以及對應的split。

為Mapper提供數據：給定某個split，能將其解析成一個個key/value對。

（2）OutputFormat：用於描述輸出數據的格式，它能夠將用戶提供的key/value對寫入特定格式的文件中。

（3）Mapper/Reducer: Mapper/Reducer中封裝了應用程序的數據處理邏輯。

（4）Writable:Hadoop自定義的序列化接口。實現該類的接口可以用作MapReduce過程中的value數據使用。

（5）WritableComparable：在Writable基礎上繼承了Comparable接口，實現該類的接口可以用作MapReduce過程中的key數據使用。（因為key包含了比較排序的操作）。

二 單詞計數實驗

　　！單詞計數文件word

　　　　

　　1‘　　啟動Hadoop　　執行命令啟動（前面博客）部署好的Hadoop系統。

　　　　　　　　命令：　　

　　　　　　　　　　　　　　cd /usr/cstor/hadoop/

　　　　　　　　　　　　　　 sbin/start-all.sh

　　　　　　　　

　　2’　　驗證HDFS上沒有wordcount的文件夾　　此時HDFS上應該是沒有wordcount文件夾。

　　　　　　　　　　　　　　　　 cd /usr/cstor/hadoop/

　　　　　　　　　　　　　　　　 bin/hadoop fs -ls /                    #查看HDFS上根目錄文件 /

　　　　　　　　

　　3‘　　上傳數據文件到HDFS

 　　　　　　　　　　　　　　cd /usr/cstor/hadoop/

 　　　　　　　　　　　　　　bin/hadoop fs -put /root/data/5/word  /

　　　　　　　　

　　4’　　編寫MapReduce程序

　　　　　　　　在eclipse新建mapreduce項目（方法見博客>>　http://www.cnblogs.com/1996swg/p/7286136.html　），新建class類WordCount

　　　　　　　　主要編寫Map和Reduce類，其中Map過程需要繼承org.apache.hadoop.mapreduce包中Mapper類，並重寫其map方法；Reduce過程需要繼承org.apache.hadoop.mapreduce包中Reduce類，並重寫其reduce方法。　　　

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.TextInputFormat;
import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;
import org.apache.hadoop.mapreduce.lib.partition.HashPartitioner;
 
import java.io.IOException;
import java.util.StringTokenizer;
 
 
public class WordCount {
    public static class TokenizerMapper extends Mapper<Object, Text, Text, IntWritable> {
        private final static IntWritable one = new IntWritable(1);
        private Text word = new Text();
        //map方法，划分一行文本，讀一個單詞寫出一個<單詞,1>
        public void map(Object key, Text value, Context context)throws IOException, InterruptedException {
            StringTokenizer itr = new StringTokenizer(value.toString());
            while (itr.hasMoreTokens()) {
                word.set(itr.nextToken());
                context.write(word, one);//寫出<單詞,1>
            }}}
    //定義reduce類，對相同的單詞，把它們<K,VList>中的VList值全部相加
    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();//相當於<Hello,1><Hello,1>，將兩個1相加
            }
            result.set(sum);
            context.write(key, result);//寫出這個單詞，和這個單詞出現次數<單詞，單詞出現次數>
        }}
    public static void main(String[] args) throws Exception {//主方法，函數入口
        Configuration conf = new Configuration();           //實例化配置文件類
        Job job = new Job(conf, "WordCount");             //實例化Job類
        job.setInputFormatClass(TextInputFormat.class);     //指定使用默認輸入格式類
        TextInputFormat.setInputPaths(job, args[0]);      //設置待處理文件的位置
        job.setJarByClass(WordCount.class);               //設置主類名
        job.setMapperClass(TokenizerMapper.class);        //指定使用上述自定義Map類
        job.setCombinerClass(IntSumReducer.class);        //指定開啟Combiner函數
        job.setMapOutputKeyClass(Text.class);            //指定Map類輸出的<K,V>，K類型
        job.setMapOutputValueClass(IntWritable.class);     //指定Map類輸出的<K,V>，V類型
        job.setPartitionerClass(HashPartitioner.class);       //指定使用默認的HashPartitioner類
        job.setReducerClass(IntSumReducer.class);         //指定使用上述自定義Reduce類
        job.setNumReduceTasks(Integer.parseInt(args[2]));  //指定Reduce個數
        job.setOutputKeyClass(Text.class);                //指定Reduce類輸出的<K,V>,K類型
        job.setOutputValueClass(Text.class);               //指定Reduce類輸出的<K,V>,V類型
        job.setOutputFormatClass(TextOutputFormat.class);  //指定使用默認輸出格式類
        TextOutputFormat.setOutputPath(job, new Path(args[1]));    //設置輸出結果文件位置
        System.exit(job.waitForCompletion(true) ? 0 : 1);    //提交任務並監控任務狀態
    }
}

 

　5'　　打包成jar文件上傳

　　　　　　假定打包后的文件名為hdpAction.jar，主類WordCount位於包njupt下，則可使用如下命令向YARN集群提交本應用。

　　　　　　　　　　　　 ./yarn  jar  hdpAction.jar  mapreduce1.WordCount  /word  /wordcount 1

　　　　　　其中“yarn”為命令，“jar”為命令參數，后面緊跟打包后的代碼地址，“mapreduce1”為包名，“WordCount”為主類名，“/word”為輸入文件在HDFS中的位置，/wordcount為輸出文件在HDFS中的位置。

 　　　　　　注意：如果打包時明確了主類，那么在輸入命令時，就無需輸入mapreduce1.WordCount來確定主類！

 　　　　　　　　　　　　

　　　　　　結果顯示：　　

17/08/05 03:37:05 WARN util.NativeCodeLoader: Unable to load native-hadoop library for your platform... using builtin-java classes where applicable
17/08/05 03:37:06 INFO client.RMProxy: Connecting to ResourceManager at master/10.1.21.27:8032
17/08/05 03:37:06 WARN mapreduce.JobResourceUploader: Hadoop command-line option parsing not performed. Implement the Tool interface and execute your application with ToolRunner to remedy this.
17/08/05 03:37:07 INFO input.FileInputFormat: Total input paths to process : 1
17/08/05 03:37:07 INFO mapreduce.JobSubmitter: number of splits:1
17/08/05 03:37:07 INFO mapreduce.JobSubmitter: Submitting tokens for job: job_1501872322130_0004
17/08/05 03:37:07 INFO impl.YarnClientImpl: Submitted application application_1501872322130_0004
17/08/05 03:37:07 INFO mapreduce.Job: The url to track the job: http://master:8088/proxy/application_1501872322130_0004/
17/08/05 03:37:07 INFO mapreduce.Job: Running job: job_1501872322130_0004
17/08/05 03:37:12 INFO mapreduce.Job: Job job_1501872322130_0004 running in uber mode : false
17/08/05 03:37:12 INFO mapreduce.Job:  map 0% reduce 0%
17/08/05 03:37:16 INFO mapreduce.Job:  map 100% reduce 0%
17/08/05 03:37:22 INFO mapreduce.Job:  map 100% reduce 100%
17/08/05 03:37:22 INFO mapreduce.Job: Job job_1501872322130_0004 completed successfully
17/08/05 03:37:22 INFO mapreduce.Job: Counters: 49
    File System Counters
        FILE: Number of bytes read=54
        FILE: Number of bytes written=232239
        FILE: Number of read operations=0
        FILE: Number of large read operations=0
        FILE: Number of write operations=0
        HDFS: Number of bytes read=166
        HDFS: Number of bytes written=28
        HDFS: Number of read operations=6
        HDFS: Number of large read operations=0
        HDFS: Number of write operations=2
    Job Counters 
        Launched map tasks=1
        Launched reduce tasks=1
        Data-local map tasks=1
        Total time spent by all maps in occupied slots (ms)=2275
        Total time spent by all reduces in occupied slots (ms)=2598
        Total time spent by all map tasks (ms)=2275
        Total time spent by all reduce tasks (ms)=2598
        Total vcore-seconds taken by all map tasks=2275
        Total vcore-seconds taken by all reduce tasks=2598
        Total megabyte-seconds taken by all map tasks=2329600
        Total megabyte-seconds taken by all reduce tasks=2660352
    Map-Reduce Framework
        Map input records=8
        Map output records=20
        Map output bytes=154
        Map output materialized bytes=54
        Input split bytes=88
        Combine input records=20
        Combine output records=5
        Reduce input groups=5
        Reduce shuffle bytes=54
        Reduce input records=5
        Reduce output records=5
        Spilled Records=10
        Shuffled Maps =1
        Failed Shuffles=0
        Merged Map outputs=1
        GC time elapsed (ms)=47
        CPU time spent (ms)=1260
        Physical memory (bytes) snapshot=421257216
        Virtual memory (bytes) snapshot=1647611904
        Total committed heap usage (bytes)=402653184
    Shuffle Errors
        BAD_ID=0
        CONNECTION=0
        IO_ERROR=0
        WRONG_LENGTH=0
        WRONG_MAP=0
        WRONG_REDUCE=0
    File Input Format Counters 
        Bytes Read=78
    File Output Format Counters 
        Bytes Written=28

 　　　　>生成結果文件wordcount目錄下的part-r-00000，用hadoop命令查看生成文件

　　　　

三  二次排序

　　　　　　MR默認會對鍵進行排序，然而有的時候我們也有對值進行排序的需求。滿足這種需求一是可以在reduce階段排序收集過來的values，但是，如果有數量巨大的values可能就會導致內存溢出等問題，這就是二次排序應用的場景——將對值的排序也安排到MR計算過程之中，而不是單獨來做。

　　　　　　二次排序就是首先按照第一字段排序，然后再對第一字段相同的行按照第二字段排序，注意不能破壞第一次排序的結果。

　　　　　　!需排序文件secsortdata.txt

　　　　　　

1'　　編寫程序IntPair 類和主類 SecondarySort類

　　　　同第一個實驗在eclipse編程的創建方法！

　　　　程序主要難點在於排序和聚合。

　　　　　　對於排序我們需要定義一個IntPair類用於數據的存儲，並在IntPair類內部自定義Comparator類以實現第一字段和第二字段的比較。

　　　　　　對於聚合我們需要定義一個FirstPartitioner類，在FirstPartitioner類內部指定聚合規則為第一字段。

　　　　　　此外，我們還需要開啟MapReduce框架自定義Partitioner 功能和GroupingComparator功能。

　　　　Inpair.java

package mr;
 
import java.io.DataInput;
import java.io.DataOutput;
import java.io.IOException;
 
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.WritableComparable;
 
public class IntPair implements WritableComparable<IntPair> {
    private IntWritable first;
    private IntWritable second;
    public void set(IntWritable first, IntWritable second) {
        this.first = first;
        this.second = second;
    }
    //注意：需要添加無參的構造方法，否則反射時會報錯。
    public IntPair() {
        set(new IntWritable(), new IntWritable());
    }
    public IntPair(int first, int second) {
        set(new IntWritable(first), new IntWritable(second));
    }
    public IntPair(IntWritable first, IntWritable second) {
        set(first, second);
    }
    public IntWritable getFirst() {
        return first;
    }
    public void setFirst(IntWritable first) {
        this.first = first;
    }
    public IntWritable getSecond() {
        return second;
    }
    public void setSecond(IntWritable second) {
        this.second = second;
    }
    public void write(DataOutput out) throws IOException {
        first.write(out);
        second.write(out);
    }
    public void readFields(DataInput in) throws IOException {
        first.readFields(in);
        second.readFields(in);
    }
    public int hashCode() {
        return first.hashCode() * 163 + second.hashCode();
    }
    public boolean equals(Object o) {
        if (o instanceof IntPair) {
            IntPair tp = (IntPair) o;
            return first.equals(tp.first) && second.equals(tp.second);
        }
        return false;
    }
    public String toString() {
        return first + "\t" + second;
    }
    public int compareTo(IntPair tp) {
        int cmp = first.compareTo(tp.first);
        if (cmp != 0) {
            return cmp;
        }
        return second.compareTo(tp.second);
    }
}

 

　　　　secsortdata.java

package mr;
 
import java.io.IOException;
 
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.io.WritableComparable;
import org.apache.hadoop.io.WritableComparator;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Partitioner;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
 
public class SecondarySort {
    static class TheMapper extends Mapper<LongWritable, Text, IntPair, NullWritable> {
        @Override
        protected void map(LongWritable key, Text value, Context context)
                throws IOException, InterruptedException {
            String[] fields = value.toString().split("\t");
            int field1 = Integer.parseInt(fields[0]);
            int field2 = Integer.parseInt(fields[1]);
            context.write(new IntPair(field1,field2), NullWritable.get());
        }
    }
    static class TheReducer extends Reducer<IntPair, NullWritable,IntPair, NullWritable> {
        //private static final Text SEPARATOR = new Text("------------------------------------------------");
        @Override
        protected void reduce(IntPair key, Iterable<NullWritable> values, Context context)
                throws IOException, InterruptedException {
            context.write(key, NullWritable.get());
        }
    }
    public static class FirstPartitioner extends Partitioner<IntPair, NullWritable> {
        public int getPartition(IntPair key, NullWritable value,
                int numPartitions) {
            return Math.abs(key.getFirst().get()) % numPartitions;
        }
    }
    //如果不添加這個類，默認第一列和第二列都是升序排序的。
//這個類的作用是使第一列升序排序，第二列降序排序
    public static class KeyComparator extends WritableComparator {
        //無參構造器必須加上，否則報錯。
        protected KeyComparator() {
            super(IntPair.class, true);
        }
        public int compare(WritableComparable a, WritableComparable b) {
            IntPair ip1 = (IntPair) a;
            IntPair ip2 = (IntPair) b;
            //第一列按升序排序
            int cmp = ip1.getFirst().compareTo(ip2.getFirst());
            if (cmp != 0) {
                return cmp;
            }
            //在第一列相等的情況下，第二列按倒序排序
            return -ip1.getSecond().compareTo(ip2.getSecond());
        }
    }
    //入口程序
    public static void main(String[] args) throws Exception {
        Configuration conf = new Configuration();
        Job job = Job.getInstance(conf);
        job.setJarByClass(SecondarySort.class);
        //設置Mapper的相關屬性
        job.setMapperClass(TheMapper.class);
        //當Mapper中的輸出的key和value的類型和Reduce輸出
//的key和value的類型相同時，以下兩句可以省略。
        //job.setMapOutputKeyClass(IntPair.class);
        //job.setMapOutputValueClass(NullWritable.class);
        FileInputFormat.setInputPaths(job, new Path(args[0]));
        //設置分區的相關屬性
        job.setPartitionerClass(FirstPartitioner.class);
        //在map中對key進行排序
        job.setSortComparatorClass(KeyComparator.class);
        //job.setGroupingComparatorClass(GroupComparator.class);
        //設置Reducer的相關屬性
        job.setReducerClass(TheReducer.class);
        job.setOutputKeyClass(IntPair.class);
        job.setOutputValueClass(NullWritable.class);
        FileOutputFormat.setOutputPath(job, new Path(args[1]));
        //設置Reducer數量
        int reduceNum = 1;
        if(args.length >= 3 && args[2] != null){
            reduceNum = Integer.parseInt(args[2]);
        }
        job.setNumReduceTasks(reduceNum);
        job.waitForCompletion(true);
    }
}

 

　2’　　打包提交

　　　　　　使用Eclipse開發工具將該代碼打包，選擇主類為mr.Secondary。如果沒有指定主類，那么在執行時就要指定須執行的類。假定打包后的文件名為Secondary.jar，主類SecondarySort位於包mr下，則可使用如下命令向Hadoop集群提交本應用。

　　　　　　 bin/hadoop jar hdpAction6.jar mr.Secondary /user/mapreduce/secsort/in/secsortdata.txt  /user/mapreduce/secsort/out  1

　　　　　　其中“hadoop”為命令，“jar”為命令參數，后面緊跟打的包，/user/mapreduce/secsort/in/secsortdata.txt”為輸入文件在HDFS中的位置，如果HDFS中沒有這個文件，則自己自行上傳。“/user/mapreduce/secsort/out/”為輸出文件在HDFS中的位置，“1”為Reduce個數。

　　　　　　如果打包時已經設定了主類，此時命令中無需再次輸入定義主類！

　　　　　　(上傳secsortdata.txt到HDFS　　命令：　”　hadoop fs -put  目標文件包括路徑  hdfs路徑  “)

　　　　　　

　　　　　顯示結果：

[root@master hadoop]# bin/hadoop jar SecondarySort.jar  /secsortdata.txt  /user/mapreduce/secsort/out  1
Not a valid JAR: /usr/cstor/hadoop/SecondarySort.jar
[root@master hadoop]# bin/hadoop jar hdpAction6.jar  /secsortdata.txt  /user/mapreduce/secsort/out  1
17/08/05 04:05:48 WARN util.NativeCodeLoader: Unable to load native-hadoop library for your platform... using builtin-java classes where applicable
17/08/05 04:05:49 INFO client.RMProxy: Connecting to ResourceManager at master/10.1.21.27:8032
17/08/05 04:05:49 WARN mapreduce.JobResourceUploader: Hadoop command-line option parsing not performed. Implement the Tool interface and execute your application with ToolRunner to remedy this.
17/08/05 04:05:50 INFO input.FileInputFormat: Total input paths to process : 1
17/08/05 04:05:50 INFO mapreduce.JobSubmitter: number of splits:1
17/08/05 04:05:50 INFO mapreduce.JobSubmitter: Submitting tokens for job: job_1501872322130_0007
17/08/05 04:05:50 INFO impl.YarnClientImpl: Submitted application application_1501872322130_0007
17/08/05 04:05:50 INFO mapreduce.Job: The url to track the job: http://master:8088/proxy/application_1501872322130_0007/
17/08/05 04:05:50 INFO mapreduce.Job: Running job: job_1501872322130_0007
17/08/05 04:05:56 INFO mapreduce.Job: Job job_1501872322130_0007 running in uber mode : false
17/08/05 04:05:56 INFO mapreduce.Job:  map 0% reduce 0%
17/08/05 04:06:00 INFO mapreduce.Job:  map 100% reduce 0%
17/08/05 04:06:05 INFO mapreduce.Job:  map 100% reduce 100%
17/08/05 04:06:06 INFO mapreduce.Job: Job job_1501872322130_0007 completed successfully
17/08/05 04:06:07 INFO mapreduce.Job: Counters: 49
    File System Counters
        FILE: Number of bytes read=106
        FILE: Number of bytes written=230897
        FILE: Number of read operations=0
        FILE: Number of large read operations=0
        FILE: Number of write operations=0
        HDFS: Number of bytes read=159
        HDFS: Number of bytes written=60
        HDFS: Number of read operations=6
        HDFS: Number of large read operations=0
        HDFS: Number of write operations=2
    Job Counters 
        Launched map tasks=1
        Launched reduce tasks=1
        Data-local map tasks=1
        Total time spent by all maps in occupied slots (ms)=2534
        Total time spent by all reduces in occupied slots (ms)=2799
        Total time spent by all map tasks (ms)=2534
        Total time spent by all reduce tasks (ms)=2799
        Total vcore-seconds taken by all map tasks=2534
        Total vcore-seconds taken by all reduce tasks=2799
        Total megabyte-seconds taken by all map tasks=2594816
        Total megabyte-seconds taken by all reduce tasks=2866176
    Map-Reduce Framework
        Map input records=10
        Map output records=10
        Map output bytes=80
        Map output materialized bytes=106
        Input split bytes=99
        Combine input records=0
        Combine output records=0
        Reduce input groups=10
        Reduce shuffle bytes=106
        Reduce input records=10
        Reduce output records=10
        Spilled Records=20
        Shuffled Maps =1
        Failed Shuffles=0
        Merged Map outputs=1
        GC time elapsed (ms)=55
        CPU time spent (ms)=1490
        Physical memory (bytes) snapshot=419209216
        Virtual memory (bytes) snapshot=1642618880
        Total committed heap usage (bytes)=402653184
    Shuffle Errors
        BAD_ID=0
        CONNECTION=0
        IO_ERROR=0
        WRONG_LENGTH=0
        WRONG_MAP=0
        WRONG_REDUCE=0
    File Input Format Counters 
        Bytes Read=60
    File Output Format Counters 
        Bytes Written=60

　　　生成文件顯示二次排序結果：

　　　　

四  計數器

1‘　　MapReduce計數器是什么？                                                                

　　　　計數器是用來記錄job的執行進度和狀態的。它的作用可以理解為日志。我們可以在程序的某個位置插入計數器，記錄數據或者進度的變化情況。

　　　MapReduce計數器能做什么？

　　　　MapReduce 計數器（Counter）為我們提供一個窗口，用於觀察 MapReduce Job 運行期的各種細節數據。對MapReduce性能調優很有幫助，MapReduce性能優化的評估大部分都是基於這些 Counter 的數值表現出來的。

　　　　在許多情況下，一個用戶需要了解待分析的數據，盡管這並非所要執行的分析任務 的核心內容。以統計數據集中無效記錄數目的任務為例，如果發現無效記錄的比例 相當高，那么就需要認真思考為何存在如此多無效記錄。是所采用的檢測程序存在 缺陷，還是數據集質量確實很低，包含大量無效記錄？如果確定是數據集的質量問 題，則可能需要擴大數據集的規模，以增大有效記錄的比例，從而進行有意義的分析。 
    　　　計數器是一種收集作業統計信息的有效手段，用於質量控制或應用級統計。計數器 還可輔助診斷系統故障。如果需要將日志信息傳輸到map或reduce任務，更好的 方法通常是嘗試傳輸計數器值以監測某一特定事件是否發生。對於大型分布式作業 而言，使用計數器更為方便。首先，獲取計數器值比輸出日志更方便，其次，根據 計數器值統計特定事件的發生次數要比分析一堆日志文件容易得多。 

　　內置計數器

 　　　　MapReduce 自帶了許多默認Counter，現在我們來分析這些默認 Counter 的含義，方便大家觀察 Job 結果，如輸入的字節數、輸出的字節數、Map端輸入/輸出的字節數和條數、Reduce端的輸入/輸出的字節數和條數等。下面我們只需了解這些內置計數器，知道計數器組名稱（groupName）和計數器名稱（counterName），以后使用計數器會查找groupName和counterName即可。　　

　　自定義計數器

　　　　MapReduce允許用戶編寫程序來定義計數器，計數器的值可在mapper或reducer 中增加。多個計數器由一個Java枚舉(enum)類型來定義，以便對計數器分組。一個作業可以定義的枚舉類型數量不限，各個枚舉類型所包含的字段數量也不限。枚 舉類型的名稱即為組的名稱，枚舉類型的字段就是計數器名稱。計數器是全局的。換言之，MapReduce框架將跨所有map和reduce聚集這些計數器，並在作業結束 時產生一個最終結果。

2’　　>編輯計數文件counters.txt

　　　　

　　　>上傳該文件到HDFS

　　　　

3'　　編寫程序Counters.java

package mr ;
import java.io.IOException;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Counter;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.util.GenericOptionsParser;
 
public class Counters {
public static class MyCounterMap extends Mapper<LongWritable, Text, Text, Text> {
           public static Counter ct = null;
           protected void map(LongWritable key, Text value,
                             org.apache.hadoop.mapreduce.Mapper<LongWritable, Text, Text, Text>.Context context)
                             throws java.io.IOException, InterruptedException {
                    String arr_value[] = value.toString().split("\t");
if (arr_value.length > 3) {
ct = context.getCounter("ErrorCounter", "toolong"); // ErrorCounter為組名，toolong為組員名
ct.increment(1); // 計數器加一
} else if (arr_value.length < 3) {
ct = context.getCounter("ErrorCounter", "tooshort");
                    ct.increment(1);
           }
}
}
public static void main(String[] args) throws IOException, InterruptedException, ClassNotFoundException {
Configuration conf = new Configuration();
String[] otherArgs = new GenericOptionsParser(conf, args).getRemainingArgs();
if (otherArgs.length != 2) {
           System.err.println("Usage: Counters <in> <out>");
System.exit(2);
}
Job job = new Job(conf, "Counter");
           job.setJarByClass(Counters.class);
 
           job.setMapperClass(MyCounterMap.class);
 
           FileInputFormat.addInputPath(job, new Path(otherArgs[0]));
           FileOutputFormat.setOutputPath(job, new Path(otherArgs[1]));
           System.exit(job.waitForCompletion(true) ? 0 : 1);
}
}

 

4'　　打包並提交

使用Eclipse開發工具將該代碼打包，選擇主類為mr.Counters。假定打包后的文件名為hdpAction7.jar，主類Counters位於包mr下，則可使用如下命令向Hadoop集群提交本應用。

　　　　　　　　bin/hadoop  jar  hdpAction7.jar  mr.Counters   /counters.txt  /usr/counters/out

其中“hadoop”為命令，“jar”為命令參數，后面緊跟打包。 “/usr/counts/in/counts.txt”為輸入文件在HDFS中的位置(如果沒有，自行上傳)，“/usr/counts/out”為輸出文件在HDFS中的位置。

 　　　顯示結果：　　　　

[root@master hadoop]# bin/hadoop  jar  hdpAction7.jar   /counters.txt  /usr/counters/out
17/08/05 04:22:58 WARN util.NativeCodeLoader: Unable to load native-hadoop library for your platform... using builtin-java classes where applicable
17/08/05 04:22:59 INFO client.RMProxy: Connecting to ResourceManager at master/10.1.21.27:8032
17/08/05 04:23:00 INFO input.FileInputFormat: Total input paths to process : 1
17/08/05 04:23:00 INFO mapreduce.JobSubmitter: number of splits:1
17/08/05 04:23:00 INFO mapreduce.JobSubmitter: Submitting tokens for job: job_1501872322130_0008
17/08/05 04:23:00 INFO impl.YarnClientImpl: Submitted application application_1501872322130_0008
17/08/05 04:23:00 INFO mapreduce.Job: The url to track the job: http://master:8088/proxy/application_1501872322130_0008/
17/08/05 04:23:00 INFO mapreduce.Job: Running job: job_1501872322130_0008
17/08/05 04:23:05 INFO mapreduce.Job: Job job_1501872322130_0008 running in uber mode : false
17/08/05 04:23:05 INFO mapreduce.Job:  map 0% reduce 0%
17/08/05 04:23:10 INFO mapreduce.Job:  map 100% reduce 0%
17/08/05 04:23:15 INFO mapreduce.Job:  map 100% reduce 100%
17/08/05 04:23:16 INFO mapreduce.Job: Job job_1501872322130_0008 completed successfully
17/08/05 04:23:16 INFO mapreduce.Job: Counters: 51
    File System Counters
        FILE: Number of bytes read=6
        FILE: Number of bytes written=229309
        FILE: Number of read operations=0
        FILE: Number of large read operations=0
        FILE: Number of write operations=0
        HDFS: Number of bytes read=134
        HDFS: Number of bytes written=0
        HDFS: Number of read operations=6
        HDFS: Number of large read operations=0
        HDFS: Number of write operations=2
    Job Counters 
        Launched map tasks=1
        Launched reduce tasks=1
        Data-local map tasks=1
        Total time spent by all maps in occupied slots (ms)=2400
        Total time spent by all reduces in occupied slots (ms)=2472
        Total time spent by all map tasks (ms)=2400
        Total time spent by all reduce tasks (ms)=2472
        Total vcore-seconds taken by all map tasks=2400
        Total vcore-seconds taken by all reduce tasks=2472
        Total megabyte-seconds taken by all map tasks=2457600
        Total megabyte-seconds taken by all reduce tasks=2531328
    Map-Reduce Framework
        Map input records=4
        Map output records=0
        Map output bytes=0
        Map output materialized bytes=6
        Input split bytes=96
        Combine input records=0
        Combine output records=0
        Reduce input groups=0
        Reduce shuffle bytes=6
        Reduce input records=0
        Reduce output records=0
        Spilled Records=0
        Shuffled Maps =1
        Failed Shuffles=0
        Merged Map outputs=1
        GC time elapsed (ms)=143
        CPU time spent (ms)=1680
        Physical memory (bytes) snapshot=413036544
        Virtual memory (bytes) snapshot=1630470144
        Total committed heap usage (bytes)=402653184
    ErrorCounter
        toolong=1
        tooshort=1
    Shuffle Errors
        BAD_ID=0
        CONNECTION=0
        IO_ERROR=0
        WRONG_LENGTH=0
        WRONG_MAP=0
        WRONG_REDUCE=0
    File Input Format Counters 
        Bytes Read=38
    File Output Format Counters 
        Bytes Written=0
[root@master hadoop]# 

 

 

五  join操作

1'　　概述

　　　　對於RDBMS中的Join操作大伙一定非常熟悉，寫SQL的時候要十分注意細節，稍有差池就會耗時巨久造成很大的性能瓶頸，而在Hadoop中使用MapReduce框架進行Join的操作時同樣耗時，但是由於Hadoop的分布式設計理念的特殊性，因此對於這種Join操作同樣也具備了一定的特殊性。

　　   原理

使用MapReduce實現Join操作有多種實現方式：

>在Reduce端連接為最為常見的模式：

Map端的主要工作：為來自不同表(文件)的key/value對打標簽以區別不同來源的記錄。然后用連接字段作為key，其余部分和新加的標志作為value，最后進行輸出。

Reduce端的主要工作：在Reduce端以連接字段作為key的分組已經完成，我們只需要在每一個分組當中將那些來源於不同文件的記錄(在map階段已經打標志)分開，最后進行笛卡爾只就OK了。

>在Map端進行連接

使用場景：一張表十分小、一張表很大。

用法:在提交作業的時候先將小表文件放到該作業的DistributedCache中，然后從DistributeCache中取出該小表進行Join key / value解釋分割放到內存中(可以放大Hash Map等等容器中)。然后掃描大表，看大表中的每條記錄的Join key /value值是否能夠在內存中找到相同Join key的記錄，如果有則直接輸出結果。

>SemiJoin

SemiJoin就是所謂的半連接，其實仔細一看就是Reduce Join的一個變種，就是在map端過濾掉一些數據，在網絡中只傳輸參與連接的數據不參與連接的數據不必在網絡中進行傳輸，從而減少了shuffle的網絡傳輸量，使整體效率得到提高，其他思想和Reduce Join是一模一樣的。說得更加接地氣一點就是將小表中參與Join的key單獨抽出來通過DistributedCach分發到相關節點，然后將其取出放到內存中(可以放到HashSet中)，在map階段掃描連接表，將Join key不在內存HashSet中的記錄過濾掉，讓那些參與Join的記錄通過shuffle傳輸到Reduce端進行Join操作，其他的和Reduce Join都是一樣的

2'　　>創建兩個表文件data.txt 和 info.txt

　　　　

　　　　>上傳到HDFS

　　　　

 

3‘　　編寫程序MRJoin.java

　　　　程序分析執行過程如下：

在map階段，把所有記錄標記成<key, value>的形式，其中key是1003/1004/1005/1006的字段值，value則根據來源不同取不同的形式：來源於表A的記錄，value的值為“201001 abc”等值；來源於表B的記錄，value的值為”kaka“之類的值。

在Reduce階段，先把每個key下的value列表拆分為分別來自表A和表B的兩部分，分別放入兩個向量中。然后遍歷兩個向量做笛卡爾積，形成一條條最終結果。

　　　　代碼如下：

package mr;
 
import java.io.DataInput;
import java.io.DataOutput;
import java.io.IOException;
 
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.io.WritableComparable;
import org.apache.hadoop.io.WritableComparator;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Partitioner;
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.lib.input.FileSplit;
import org.apache.hadoop.util.GenericOptionsParser;
 
public class MRJoin {
public static class MR_Join_Mapper extends Mapper<LongWritable, Text, TextPair, Text> {
         @Override
           protected void map(LongWritable key, Text value, Context context)
                                                                                             throws IOException, InterruptedException {
                    // 獲取輸入文件的全路徑和名稱
                    String pathName = ((FileSplit) context.getInputSplit()).getPath().toString();
                    if (pathName.contains("data.txt")) {
                             String values[] = value.toString().split("\t");
                             if (values.length < 3) {
                                       // data數據格式不規范，字段小於3，拋棄數據
                                       return;
                             } else {
                                       // 數據格式規范，區分標識為1
                                       TextPair tp = new TextPair(new Text(values[1]), new Text("1"));
                                       context.write(tp, new Text(values[0] + "\t" + values[2]));
                             }
                    }
                    if (pathName.contains("info.txt")) {
                             String values[] = value.toString().split("\t");
                             if (values.length < 2) {
                                       // data數據格式不規范，字段小於2，拋棄數據
                                       return;
                             } else {
                                       // 數據格式規范，區分標識為0
                                       TextPair tp = new TextPair(new Text(values[0]), new Text("0"));
                                       context.write(tp, new Text(values[1]));
                             }
                    }
           }
}
 
public static class MR_Join_Partitioner extends Partitioner<TextPair, Text> {
           @Override
           public int getPartition(TextPair key, Text value, int numParititon) {
                    return Math.abs(key.getFirst().hashCode() * 127) % numParititon;
           }
}
 
public static class MR_Join_Comparator extends WritableComparator {
           public MR_Join_Comparator() {
                    super(TextPair.class, true);
           }
 
           public int compare(WritableComparable a, WritableComparable b) {
                    TextPair t1 = (TextPair) a;
                    TextPair t2 = (TextPair) b;
                    return t1.getFirst().compareTo(t2.getFirst());
           }
}
 
public static class MR_Join_Reduce extends Reducer<TextPair, Text, Text, Text> {
           protected void Reduce(TextPair key, Iterable<Text> values, Context context)
                             throws IOException, InterruptedException {
                    Text pid = key.getFirst();
                    String desc = values.iterator().next().toString();
                    while (values.iterator().hasNext()) {
                             context.write(pid, new Text(values.iterator().next().toString() + "\t" + desc));
                    }
           }
}
 
 
public static void main(String agrs[])
                                                throws IOException, InterruptedException, ClassNotFoundException {
           Configuration conf = new Configuration();
           GenericOptionsParser parser = new GenericOptionsParser(conf, agrs);
           String[] otherArgs = parser.getRemainingArgs();
           if (agrs.length < 3) {
                    System.err.println("Usage: MRJoin <in_path_one> <in_path_two> <output>");
                    System.exit(2);
           }
          
           Job job = new Job(conf, "MRJoin");
           // 設置運行的job
           job.setJarByClass(MRJoin.class);
           // 設置Map相關內容
           job.setMapperClass(MR_Join_Mapper.class);
           // 設置Map的輸出
           job.setMapOutputKeyClass(TextPair.class);
           job.setMapOutputValueClass(Text.class);
           // 設置partition
           job.setPartitionerClass(MR_Join_Partitioner.class);
           // 在分區之后按照指定的條件分組
           job.setGroupingComparatorClass(MR_Join_Comparator.class);
           // 設置Reduce
           job.setReducerClass(MR_Join_Reduce.class);
           // 設置Reduce的輸出
           job.setOutputKeyClass(Text.class);
           job.setOutputValueClass(Text.class);
           // 設置輸入和輸出的目錄
           FileInputFormat.addInputPath(job, new Path(otherArgs[0]));
           FileInputFormat.addInputPath(job, new Path(otherArgs[1]));
           FileOutputFormat.setOutputPath(job, new Path(otherArgs[2]));
           // 執行，直到結束就退出
           System.exit(job.waitForCompletion(true) ? 0 : 1);
 }
}
 
class TextPair implements WritableComparable<TextPair> {
private Text first;
private Text second;
 
public TextPair() {
           set(new Text(), new Text());
}
 
public TextPair(String first, String second) {
           set(new Text(first), new Text(second));
}
 
public TextPair(Text first, Text second) {
           set(first, second);
}
 
public void set(Text first, Text second) {
           this.first = first;
           this.second = second;
}
 
public Text getFirst() {
           return first;
}
 
public Text getSecond() {
           return second;
}
 
public void write(DataOutput out) throws IOException {
           first.write(out);
           second.write(out);
}
 
public void readFields(DataInput in) throws IOException {
           first.readFields(in);
           second.readFields(in);
}
 
public int compareTo(TextPair tp) {
           int cmp = first.compareTo(tp.first);
           if (cmp != 0) {
                    return cmp;
           }
           return second.compareTo(tp.second);
}
}

 

 4’　　打包並提交

使用Eclipse開發工具將該代碼打包，假定打包后的文件名為hdpAction8.jar，主類MRJoin位於包mr下，則可使用如下命令向Hadoop集群提交本應用。

　　　　　　 bin/hadoop jar hdpAction8.jar mr.MRJoin  /data.txt    /info.txt   /usr/MRJoin/out

其中“hadoop”為命令，“jar”為命令參數，后面緊跟打包。 “/data.txt”和 “/info.txt”為輸入文件在HDFS中的位置，“/usr/MRJoin/out”為輸出文件在HDFS中的位置。

　　執行結果如下：　　　

[root@master hadoop]# bin/hadoop jar hdpAction8.jar  /data.txt    /info.txt   /usr/MRJoin/out
17/08/05 04:38:11 WARN util.NativeCodeLoader: Unable to load native-hadoop library for your platform... using builtin-java classes where applicable
17/08/05 04:38:12 INFO client.RMProxy: Connecting to ResourceManager at master/10.1.21.27:8032
17/08/05 04:38:13 INFO input.FileInputFormat: Total input paths to process : 2
17/08/05 04:38:13 INFO mapreduce.JobSubmitter: number of splits:2
17/08/05 04:38:13 INFO mapreduce.JobSubmitter: Submitting tokens for job: job_1501872322130_0009
17/08/05 04:38:13 INFO impl.YarnClientImpl: Submitted application application_1501872322130_0009
17/08/05 04:38:13 INFO mapreduce.Job: The url to track the job: http://master:8088/proxy/application_1501872322130_0009/
17/08/05 04:38:13 INFO mapreduce.Job: Running job: job_1501872322130_0009
17/08/05 04:38:18 INFO mapreduce.Job: Job job_1501872322130_0009 running in uber mode : false
17/08/05 04:38:18 INFO mapreduce.Job:  map 0% reduce 0%
17/08/05 04:38:23 INFO mapreduce.Job:  map 100% reduce 0%
17/08/05 04:38:28 INFO mapreduce.Job:  map 100% reduce 100%
17/08/05 04:38:28 INFO mapreduce.Job: Job job_1501872322130_0009 completed successfully
17/08/05 04:38:29 INFO mapreduce.Job: Counters: 49
    File System Counters
        FILE: Number of bytes read=179
        FILE: Number of bytes written=347823
        FILE: Number of read operations=0
        FILE: Number of large read operations=0
        FILE: Number of write operations=0
        HDFS: Number of bytes read=317
        HDFS: Number of bytes written=283
        HDFS: Number of read operations=9
        HDFS: Number of large read operations=0
        HDFS: Number of write operations=2
    Job Counters 
        Launched map tasks=2
        Launched reduce tasks=1
        Data-local map tasks=2
        Total time spent by all maps in occupied slots (ms)=5122
        Total time spent by all reduces in occupied slots (ms)=2685
        Total time spent by all map tasks (ms)=5122
        Total time spent by all reduce tasks (ms)=2685
        Total vcore-seconds taken by all map tasks=5122
        Total vcore-seconds taken by all reduce tasks=2685
        Total megabyte-seconds taken by all map tasks=5244928
        Total megabyte-seconds taken by all reduce tasks=2749440
    Map-Reduce Framework
        Map input records=10
        Map output records=10
        Map output bytes=153
        Map output materialized bytes=185
        Input split bytes=184
        Combine input records=0
        Combine output records=0
        Reduce input groups=4
        Reduce shuffle bytes=185
        Reduce input records=10
        Reduce output records=10
        Spilled Records=20
        Shuffled Maps =2
        Failed Shuffles=0
        Merged Map outputs=2
        GC time elapsed (ms)=122
        CPU time spent (ms)=2790
        Physical memory (bytes) snapshot=680472576
        Virtual memory (bytes) snapshot=2443010048
        Total committed heap usage (bytes)=603979776
    Shuffle Errors
        BAD_ID=0
        CONNECTION=0
        IO_ERROR=0
        WRONG_LENGTH=0
        WRONG_MAP=0
        WRONG_REDUCE=0
    File Input Format Counters 
        Bytes Read=133
    File Output Format Counters 
        Bytes Written=283
[root@master hadoop]# 

 　　　>　生成join后的文件在/usr/MRJoin/out目錄下：

　　

六  分布式緩存

1‘ 　假定現有一個大為100G的大表big.txt和一個大小為1M的小表small.txt，請基於MapReduce思想編程實現判斷小表中單詞在大表中出現次數。也即所謂的“掃描大表、加載小表”。

　　為解決上述問題，可開啟10個Map。這樣，每個Map只需處理總量的1/10，將大大加快處理。而在單獨Map內，直接用HashSet加載“1M小表”，對於存在硬盤（Map處理時會將HDFS文件拷貝至本地）的10G大文件，則逐條掃描，這就是所謂的“掃描大表、加載小表”，也即分布式緩存。

2’　　>新建兩個txt文件

　　　　

　　　>上傳到HDFS

　　　　首先登錄client機，查看HDFS里是否已存在目錄“/user/root/mr/in”，若不存在，使用下述命令新建該目錄。

　　　　　　　　 /usr/cstor/hadoop/bin/hdfs  dfs  -mkdir  -p  /user/root/mr/in

　　　　　　　　

　　　　接着，使用下述命令將client機本地文件“/root/data/9/big.txt”和“/root/data/9/ small.txt”上傳至HDFS的“/user/root/mr/in”目錄：

　　　　　　　　 /usr/cstor/hadoop/bin/hdfs  dfs  -put  /root/data/9/big.txt  /user/root/mr/in

　　　　　　　　 /usr/cstor/hadoop/bin/hdfs  dfs  -put  /root/data/9/small.txt  /user/root/mr/in

　　　　　　　　

3‘　　編寫代碼，新建BigAndSmallTable類並指定包名（代碼中為cn.cstor.mr），在BigAndSmallTable.java文件中

　　　　　　依次寫入如下代碼：　　　　　　

package cn.cstor.mr;
 
import java.io.IOException;
import java.util.HashSet;
 
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FSDataInputStream;
import org.apache.hadoop.fs.FileSystem;
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.LineReader;
 
public class BigAndSmallTable {
public static class TokenizerMapper extends
Mapper<Object, Text, Text, IntWritable> {
private final static IntWritable one = new IntWritable(1);
private static HashSet<String> smallTable = null;
 
protected void setup(Context context) throws IOException,
InterruptedException {
smallTable = new HashSet<String>();
Path smallTablePath = new Path(context.getConfiguration().get(
"smallTableLocation"));
FileSystem hdfs = smallTablePath.getFileSystem(context
.getConfiguration());
FSDataInputStream hdfsReader = hdfs.open(smallTablePath);
Text line = new Text();
LineReader lineReader = new LineReader(hdfsReader);
while (lineReader.readLine(line) > 0) {
// you can do something here
String[] values = line.toString().split(" ");
for (int i = 0; i < values.length; i++) {
smallTable.add(values[i]);
System.out.println(values[i]);
}
}
lineReader.close();
hdfsReader.close();
System.out.println("setup ok *^_^* ");
}
 
public void map(Object key, Text value, Context context)
throws IOException, InterruptedException {
String[] values = value.toString().split(" ");
for (int i = 0; i < values.length; i++) {
if (smallTable.contains(values[i])) {
context.write(new Text(values[i]), one);
}
}
}
}
 
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();
conf.set("smallTableLocation", args[1]);
Job job = Job.getInstance(conf, "BigAndSmallTable");
job.setJarByClass(BigAndSmallTable.class);
job.setMapperClass(TokenizerMapper.class);
job.setReducerClass(IntSumReducer.class);
job.setMapOutputKeyClass(Text.class);
job.setMapOutputValueClass(IntWritable.class);
job.setOutputKeyClass(Text.class);
job.setOutputValueClass(IntWritable.class);
FileInputFormat.addInputPath(job, new Path(args[0]));
FileOutputFormat.setOutputPath(job, new Path(args[2]));
System.exit(job.waitForCompletion(true) ? 0 : 1);
}
}

 4’　　打包上傳並執行　　　

　　　　首先，使用“Xmanager Enterprise 5”將“C:\Users\allen\ Desktop\BigSmallTable.jar”上傳至client機。此處上傳至“/root/BigSmallTable.jar”

　　　　接着，登錄client機上，使用下述命令提交BigSmallTable.jar任務。

　　　　　　　　 /usr/cstor/hadoop/bin/hadoop  jar  /root/BigSmallTable.jar  cn.cstor.mr.BigAndSmallTable  /user/root/mr/in/big.txt   /user/root/mr/in/small.txt  /user/root/mr/bigAndSmallResult　

[root@client ~]# /usr/cstor/hadoop/bin/hadoop  jar  /root/BigSmallTable.jar   /user/root/mr/in/big.txt   /user/root/mr/in/small.txt  /user/root/mr/bigAndSmallResult
17/08/05 04:55:51 INFO client.RMProxy: Connecting to ResourceManager at master/10.1.21.27:8032
17/08/05 04:55:52 WARN mapreduce.JobResourceUploader: Hadoop command-line option parsing not performed. Implement the Tool interface and execute your application with ToolRunner to remedy this.
17/08/05 04:55:52 INFO input.FileInputFormat: Total input paths to process : 1
17/08/05 04:55:52 INFO mapreduce.JobSubmitter: number of splits:1
17/08/05 04:55:52 INFO mapreduce.JobSubmitter: Submitting tokens for job: job_1501872322130_0010
17/08/05 04:55:53 INFO impl.YarnClientImpl: Submitted application application_1501872322130_0010
17/08/05 04:55:53 INFO mapreduce.Job: The url to track the job: http://master:8088/proxy/application_1501872322130_0010/
17/08/05 04:55:53 INFO mapreduce.Job: Running job: job_1501872322130_0010
17/08/05 04:55:58 INFO mapreduce.Job: Job job_1501872322130_0010 running in uber mode : false
17/08/05 04:55:58 INFO mapreduce.Job:  map 0% reduce 0%
17/08/05 04:56:03 INFO mapreduce.Job:  map 100% reduce 0%
17/08/05 04:56:08 INFO mapreduce.Job:  map 100% reduce 100%
17/08/05 04:56:09 INFO mapreduce.Job: Job job_1501872322130_0010 completed successfully
17/08/05 04:56:09 INFO mapreduce.Job: Counters: 49
    File System Counters
        FILE: Number of bytes read=36
        FILE: Number of bytes written=231153
        FILE: Number of read operations=0
        FILE: Number of large read operations=0
        FILE: Number of write operations=0
        HDFS: Number of bytes read=265
        HDFS: Number of bytes written=18
        HDFS: Number of read operations=7
        HDFS: Number of large read operations=0
        HDFS: Number of write operations=2
    Job Counters 
        Launched map tasks=1
        Launched reduce tasks=1
        Data-local map tasks=1
        Total time spent by all maps in occupied slots (ms)=2597
        Total time spent by all reduces in occupied slots (ms)=2755
        Total time spent by all map tasks (ms)=2597
        Total time spent by all reduce tasks (ms)=2755
        Total vcore-seconds taken by all map tasks=2597
        Total vcore-seconds taken by all reduce tasks=2755
        Total megabyte-seconds taken by all map tasks=2659328
        Total megabyte-seconds taken by all reduce tasks=2821120
    Map-Reduce Framework
        Map input records=5
        Map output records=3
        Map output bytes=24
        Map output materialized bytes=36
        Input split bytes=107
        Combine input records=0
        Combine output records=0
        Reduce input groups=3
        Reduce shuffle bytes=36
        Reduce input records=3
        Reduce output records=3
        Spilled Records=6
        Shuffled Maps =1
        Failed Shuffles=0
        Merged Map outputs=1
        GC time elapsed (ms)=57
        CPU time spent (ms)=1480
        Physical memory (bytes) snapshot=425840640
        Virtual memory (bytes) snapshot=1651806208
        Total committed heap usage (bytes)=402653184
    Shuffle Errors
        BAD_ID=0
        CONNECTION=0
        IO_ERROR=0
        WRONG_LENGTH=0
        WRONG_MAP=0
        WRONG_REDUCE=0
    File Input Format Counters 
        Bytes Read=147
    File Output Format Counters 
        Bytes Written=18
[root@client ~]# 

　　　　 >查看結果

　　　　　　　　程序執行后，可使用下述命令查看執行結果，注意若再次執行，請更改結果目錄：

　　　　　　　　　　　　 /usr/cstor/hadoop/bin/hdfs  dfs  -cat  /user/root/mr/bigAndSmallResult/part-r-00000

　　　　　　

 

總結：

　　　從五個實驗做出來之后，我們可以系統化的了解mapreduce的運行流程：

首先目標文件上傳到HDFS；

其次編寫目標程序代碼；

然后將其打包上傳到集群服務器上；

再然后執行該jar包；

生成part-r-00000結果文件。

　　關於hadoop的命令使用也更加熟練，對於一些文件上傳、查看、編輯的處理也可以掌握於心了。學習到這里，對於大數據也算可以入門了，對於大數據也有了一定的了解與基本操作。

　　路漫漫其修遠兮，吾將上下而求索。日積月累，堅持不懈是學習成果的前提。量變造成質變，望看到此處的朋友們共同努力，互相交流學習，我們都是愛學習的人！