Mapreduce概述和WordCount程式

一、Mapreduce概述

　　Mapreduce是分散式程式編程框架，也是分散式計算框架，它簡化了開發！

　　Mapreduce將用戶編寫的業務邏輯代碼和自帶預設組合整合成一個完整的分散式運算程式，併發的運行在hadoop集群上。

二、Mapreduce優缺點

　　優點：1.易於編程：只用實現幾個介面即可完成一個併發的程式。

　　　　　2.良好的拓展性：再不行當前程式運行的情況下，可以通過增加節點來解決用戶/數據擴展，計算量增加的問題。

　　　　　3.高容錯性：可以運行在廉價的集群機器上。

　　　　　4.適合處理PB級別以上的離線處理。

　　缺點：1.不擅長做實時計算、流式計算。

　　　　　2.不支持DAG(有向圖)計算，有依賴的程式(spark支持)。

　　　　　3.每次把計算結果寫入磁碟當中，造成磁碟io，性能較低。

三、Mapreduce編程思想

　　需求：統計一個200M的單詞文件，查詢出每個單詞出現的次數。

　　思想：1.將200M的文件切分為兩塊，128M和72M；

　　　　　2.將兩塊文件分別交給兩個maptask處理，對數據進行讀取，切分，封裝，然後傳輸到reducetask；

　　　　　3.reducetask將數據再次整合，累加，輸出到結果文件中。

　　註意：mapreduce中的所有maptask都是並行運行的，reducetask也是，

　　　　但是reducetask的運行要依賴maptask的輸出。

四、WordCount程式

/**
 * @author: PrincessHug
 * @date: 2019/3/24, 0:52
 * @Blog: https://www.cnblogs.com/HelloBigTable/
 */
public class WordCountMapper extends Mapper<LongWritable, Text,Text, IntWritable> {
    @Override
    protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
        //讀取數據
        String line = value.toString();

        //切分數據
        String[] fields = line.split(" ");

        //傳輸數據
        for (String f:fields){
            context.write(new Text(f),new IntWritable(1));
        }
    }
}

public class WordCountReducer extends Reducer<Text, IntWritable,Text,IntWritable> {
    @Override
    protected void reduce(Text key, Iterable<IntWritable> values, Context context) throws IOException, InterruptedException {
        //累加
        int sum = 0;
        for (IntWritable i:values){
            sum += i.get();
        }
        //輸出
        context.write(key,new IntWritable(sum));
    }
}

public class WordCountDriver {
    public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException {
        //配置，job對象
        Configuration conf = new Configuration();
        Job job = Job.getInstance();

        //設置運行類
        job.setJarByClass(WordCountDriver.class);

        //設置Mapper，Reducer類
        job.setMapperClass(WordCountMapper.class);
        job.setReducerClass(WordCountReducer.class);

        //設置Mapper輸出數據類型
        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(IntWritable.class);

        //設置Reducer輸出數據類型
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(IntWritable.class);

        //設置輸入輸出流
        FileInputFormat.setInputPaths(job,new Path("G:\\mapreduce\\wordcount\\in"));
        FileOutputFormat.setOutputPath(job,new Path("G:\\mapreduce\\wordcount\\out"));

        //提交任務
        if (job.waitForCompletion(true)){
            System.out.println("運行完成！");
        }else {
            System.out.println("運行失敗！");
        }

    }

}