Hadoop中MapReduce的示例分析

這篇文章主要介紹了Hadoop中MapReduce的示例分析，具有一定借鑒價值，感興趣的朋友可以參考下，希望大家閱讀完這篇文章之后大有收獲，下面讓小編帶著大家一起了解一下。

MapReduce設計理念

• 移動計算，而不是移動數(shù)據(jù)。

MapReduce之Helloworld（Word Count）處理過程

MapReduce的Split大小 - max.split(200M) - min.split(50M) - block(128M) - max(min.split,min(max.split,block))=128M

Mapper

• Map-reducede 的 思想就是“分而治之”

• Mapper負責“分”，即把發(fā)雜的任務分解為若干個“簡單的任務”執(zhí)行

• “簡單的任務”有幾個含義：

• 數(shù)據(jù)或計算規(guī)模相對于原任務要大大縮小；

• 就近計算，即會被分配到存放了所需數(shù)據(jù)的節(jié)點進行計算；

• 這些小任務可以并行計算，彼此間幾乎沒有依賴關(guān)系。

Reduce

• 對map階段的結(jié)果進行匯總；

• reducer的數(shù)目由mapred-site.xml配置文件里的項目mapred.reduce.tasks決定。缺省值為1，用戶可以覆蓋（一般在程序中調(diào)整，不修改xml默認值）

shuffler（最為復雜的一個環(huán)節(jié)）

• 參考：MapReduce:詳解Shuffle過程

• 在mapper和reduce中間的一個步驟

• 可以把mapper的輸出按照某種key值重新切分和組合成N份，把key值符合某種范圍的組輸送到特定的reduce那里去處理

• 可以簡化reduce過程

附：Helloworld之WordCount

//WCJob.java

import java.io.IOException;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.LongWritable;
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.StringUtils;

/**
 * MapReduce_Helloworld程序
 *
 * WCJob
 * @since V1.0.0
 * Created by SET on 2016-09-11 11:35:15
 * @see
 */
public class WCJob {
    public static void main(String[] args) throws Exception {
        Configuration config = new Configuration();
        config.set("fs.defaultFS", "hdfs://master:8020");
        config.set("yarn-resourcemanager.hostname", "slave2");

        FileSystem fs = FileSystem.newInstance(config);

        Job job = new Job(config);

        job.setJobName("word count");

        job.setJarByClass(WCJob.class);

        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(IntWritable.class);

        job.setMapperClass(WCMapper.class);
        job.setReducerClass(WCReducer.class);
        
        job.setCombinerClass(WCReducer.class);

        FileInputFormat.addInputPath(job, new Path("/user/wc/wc"));
        Path outputpath = new Path("/user/wc/output");
        if(fs.exists(outputpath)) {
            fs.delete(outputpath, true);
        }
        FileOutputFormat.setOutputPath(job, outputpath);


        boolean flag = job.waitForCompletion(true);
        if(flag) {
            System.out.println("Job success@!");
        }
    }

    private static class WCMapper extends Mapper<LongWritable, Text, Text, IntWritable> {
        @Override
        protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
            /**
             * 格式：hadoop hello world
             * map 拿到每一行數(shù)據(jù) 切分
            */
            String[] strs = StringUtils.split(value.toString(), ' ');
            for(String word : strs) {
                context.write(new Text(word), new IntWritable(1));
            }
        }
    }

    private static class WCReducer 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 intWritable : values) {
                sum += intWritable.get();
            }
            context.write(new Text(key), new IntWritable(sum));
        }
    }
}

感謝你能夠認真閱讀完這篇文章，希望小編分享的“Hadoop中MapReduce的示例分析”這篇文章對大家有幫助，同時也希望大家多多支持億速云，關(guān)注億速云行業(yè)資訊頻道，更多相關(guān)知識等著你來學習!