1、MapReduce概述

1.1 MapReduce定義

MapReduce是一個分布式運算程序的編程框架，是用戶開發(fā)“基于Hadoop的數(shù)據(jù)分析應用”的核心框架。
MapReduce核心功能是將用戶編寫的業(yè)務邏輯代碼和自帶默認組件整合成一個完整的分布式運算程序，并發(fā)運行在一個Hadoop集群上。

1.2 MapReduce優(yōu)點

1）MapReduce易于編程

它簡單的實現(xiàn)一些接口，就可以完成一個分布式程序， 這個分布式程序可以分布到大量廉價的PC機器上運行。也就是說你寫一個分布式程序，跟寫一個簡單的串行程序是一模一樣的。就是因為這個特點使得MapReduce編程變得非常流行。

2）良好的擴展性

當你的計算資源不能得到滿足的時候，你可以通過簡單的增加機器來擴展它的計算能力。

3）高容錯性

MapReduce設計的初衷就是使程序能夠部署在廉價的PC機器上，這就要求它具有很高的容錯性。比如其中一臺機器掛了，它可以把上面的計算任務轉移到另外一個節(jié)點上運行，不至于這個任務運行失敗，而且這個過程不需要人工參與，而完全是由Hadoop內部完成的。

4）適合PB級以上海量數(shù)據(jù)的離線處理

可以實現(xiàn)上千臺服務器集群并發(fā)工作，提供數(shù)據(jù)處理能力。

1.3 MapReduce缺點

1）不擅長實時計算

MapReduce無法像MySQL一樣，在毫秒或者秒級內返回結果。

2）不擅長流式計算

流式計算的輸入數(shù)據(jù)是動態(tài)的，而MapReduce的輸入數(shù)據(jù)集是靜態(tài)的，不能動態(tài)變化。這是因為MapReduce自身的設計特點決定了數(shù)據(jù)源必須是靜態(tài)的。

3）不擅長DAG（有向無環(huán)圖）計算

多個應用程序存在依賴關系，后一個應用程序的輸入為前一個的輸出。在這種情況下，MapReduce并不是不能做，而是使用后，每個MapReduce作業(yè)的輸出結果都會寫入到磁盤，會造成大量的磁盤IO，導致性能非常的低下。

1.4 MapReduce核心思想

現(xiàn)在有一個需求：要統(tǒng)計一個文件當中每一個單詞出現(xiàn)的總次數(shù)（并將查詢結果a-p字母保存一個文件，q-z字母保存一個文件），則可以按照圖示步驟

（1）分布式的運算程序往往需要分成至少2個階段。map+reduce
（2）第一個階段的MapTask并發(fā)實例，完全并行運行，互不相干。統(tǒng)計次數(shù)，形成鍵值對，<H，1>、<S，1>、<H，1>，但是次數(shù)之間不相加。
（3）第二個階段的ReduceTask并發(fā)實例互不相干，但是他們的數(shù)據(jù)依賴于上一個階段的所有MapTask并發(fā)實例的輸出。將統(tǒng)計的次數(shù)相加求和。
（4）MapReduce編程模型只能包含一個Map階段和一個Reduce階段，如果用戶的業(yè)務邏輯非常復雜，那就只能多個MapReduce程序，串行運行。

總結：分析WordCount數(shù)據(jù)流走向深入理解MapReduce核心思想。

1.5 MapReduce進程

mr、job、任務指的都是一個應用程序。例如：跑一個wordcount，可以說這是一個job或者任務。

未來在運行MapReduce程序的時候，會啟動哪些進程呢？

一個完整的MapReduce程序在分布式運行時有三類實例進程：
（1）MrAppMaster：負責整個程序的過程調度及狀態(tài)協(xié)調。
（2）MapTask：負責Map階段的整個數(shù)據(jù)處理流程。
（3）ReduceTask：負責Reduce階段的整個數(shù)據(jù)處理流程。

1.6 常用數(shù)據(jù)序列化類型

• 除了string，其他的都是在java類型的基礎上加上writable

1.7 源碼與MapReduce編程規(guī)范

用戶編寫的程序分成三個部分：Mapper、Reducer和Driver。

源碼如下：

package org.apache.hadoop.examples;

import java.io.IOException;
import java.io.PrintStream;
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.Mapper.Context;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.Reducer.Context;
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 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, "word count");
    job.setJarByClass(WordCount.class);
    job.setMapperClass(TokenizerMapper.class);
    job.setCombinerClass(IntSumReducer.class);
    job.setReducerClass(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 void reduce(Text key, Iterable<IntWritable> values, Reducer<Text, IntWritable, Text, IntWritable>.Context context)
      throws IOException, InterruptedException
    {
      int sum = 0;
      for (IntWritable val : values) {
        sum += val.get();
      }
      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 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);
      }
    }
  }
}

• 上面一共有三個方法，分別是main方法，map方法和reduce方法。
• 定義一個類，繼承mapper，之后重寫里面的mapper方法，實現(xiàn)自己的業(yè)務邏輯。

MapReduce的編程規(guī)范如下：

2、WordCount案例實操

2.1 本地測試

1）需求

在給定的文本文件中統(tǒng)計輸出每一個單詞出現(xiàn)的總次數(shù)
（1）輸入數(shù)據(jù)

（2）期望輸出數(shù)據(jù)

wenxin	2
banzhang	1
cls	2
hadoop	1
jiao	1
ss	2
xue	1

• 可以發(fā)現(xiàn)上面的數(shù)據(jù)涉及首字母排序的問題。

2）需求分析

按照MapReduce編程規(guī)范，分別編寫Mapper，Reducer，Driver。

（1）創(chuàng)建maven工程，MapReduceDemo

（2）在pom.xml文件中添加如下依賴

<dependencies>
    <dependency>
        <groupId>org.apache.hadoop</groupId>
        <artifactId>hadoop-client</artifactId>
        <version>3.1.3</version>
    </dependency>
    <dependency>
        <groupId>junit</groupId>
        <artifactId>junit</artifactId>
        <version>4.12</version>
    </dependency>
    <dependency>
        <groupId>org.slf4j</groupId>
        <artifactId>slf4j-log4j12</artifactId>
        <version>1.7.30</version>
    </dependency>
</dependencies>

（2）在項目的src/main/resources目錄下，新建一個文件，命名為“l(fā)og4j.properties”，在文件中填入。

log4j.rootLogger=INFO, stdout  
log4j.appender.stdout=org.apache.log4j.ConsoleAppender  
log4j.appender.stdout.layout=org.apache.log4j.PatternLayout  
log4j.appender.stdout.layout.ConversionPattern=%d %p [%c] - %m%n  
log4j.appender.logfile=org.apache.log4j.FileAppender  
log4j.appender.logfile.File=target/spring.log  
log4j.appender.logfile.layout=org.apache.log4j.PatternLayout  
log4j.appender.logfile.layout.ConversionPattern=%d %p [%c] - %m%n

（3）創(chuàng)建包名：com.wenxin.mapreduce.wordcount

Mapper的源碼：

@Public
@Stable
public class Mapper<KEYIN, VALUEIN, KEYOUT, VALUEOUT> {
    public Mapper() {
    }

    protected void setup(Mapper<KEYIN, VALUEIN, KEYOUT, VALUEOUT>.Context context) throws IOException, InterruptedException {
    }

    protected void map(KEYIN key, VALUEIN value, Mapper<KEYIN, VALUEIN, KEYOUT, VALUEOUT>.Context context) throws IOException, InterruptedException {
        context.write(key, value);
    }

    protected void cleanup(Mapper<KEYIN, VALUEIN, KEYOUT, VALUEOUT>.Context context) throws IOException, InterruptedException {
    }

    public void run(Mapper<KEYIN, VALUEIN, KEYOUT, VALUEOUT>.Context context) throws IOException, InterruptedException {
        this.setup(context);

        try {
            while(context.nextKeyValue()) {
                this.map(context.getCurrentKey(), context.getCurrentValue(), context);
            }
        } finally {
            this.cleanup(context);
        }

    }

    public abstract class Context implements MapContext<KEYIN, VALUEIN, KEYOUT, VALUEOUT> {
        public Context() {
        }
    }
}

4）編寫程序

（1）編寫Mapper類

package com.wenxin.mapreduce.wordcount;
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;

/**
 * @author Susie-Wen
 * @version 1.0
 * @description:
 * @date 2023/12/13 9:56
 */
/*
KEYIN,map階段輸入的key的類型：LongWritable
VALUEINT,map階段輸入的value的類型：Text
KEYOUT,map階段輸出的Key的類型：Text
VALUEOUT,map階段輸出的value類型：IntWritable
 */
public class WordCountMapper<map> extends Mapper<LongWritable, Text,Text, IntWritable> {
    //
    private Text outK=new Text();
    private IntWritable outV=new IntWritable(1);
    @Override
    protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
        /*
        LongWritable key,輸入的key，偏移量
        Text value,輸入的value
        Context context,對應的上下文
         */
        //1.獲取一行
        String line = value.toString();
        
        //2.對一行數(shù)據(jù)進行切割（因為原始數(shù)據(jù)使用的是空格，因此這里使用空格切割）
        String[] words = line.split(" ");
        
        //3.循環(huán)寫出
        for(String word:words){
            //封裝outK
            outK.set(word);
            //寫出
            context.write(outK,outV);
        }
    }
}

（2）編寫Reducer類

package com.wenxin.mapreduce.wordcount;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;
import java.io.IOException;
/**
 * @author Susie-Wen
 * @version 1.0
 * @description:
 * @date 2023/12/13 9:56
 */
/*
KEYIN,reduce階段輸入的key的類型：Text
VALUEINT,reduce階段輸入的value的類型：IntWritable
KEYOUT,reduce階段輸出的Key的類型：Text
VALUEOUT,reduce階段輸出的value類型：IntWritable
 */
public class WordCountReducer extends Reducer<Text, IntWritable,Text,IntWritable> {
    IntWritable outV=new IntWritable();
    @Override
    protected void reduce(Text key, Iterable<IntWritable> values, Context context) throws IOException, InterruptedException {
        int sum =0;//定義一個變量，進行累加
        //傳進來的值：wenxin,(1,1)
        for(IntWritable value:values){
            sum +=value.get();//累加，不能直接加上value，因為value是IntWritable類型，要使用get方法
        }
        outV.set(sum);
        //寫出
        context.write(key,outV);
    }
}

（3）編寫Driver驅動類

• driver當中有7步，都是固定的；其次需要注意不要導錯包了！

package com.atguigu.mapreduce.wordcount;
import java.io.IOException;
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;

public class WordCountDriver {

	public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException {

		// 1 獲取配置信息以及獲取job對象
		Configuration conf = new Configuration();
		Job job = Job.getInstance(conf);

		// 2 關聯(lián)本Driver程序的jar
		job.setJarByClass(WordCountDriver.class);

		// 3 關聯(lián)Mapper和Reducer的jar
		job.setMapperClass(WordCountMapper.class);
		job.setReducerClass(WordCountReducer.class);

		// 4 設置Mapper輸出的kv類型
		job.setMapOutputKeyClass(Text.class);
		job.setMapOutputValueClass(IntWritable.class);

		// 5 設置最終輸出kv類型
		job.setOutputKeyClass(Text.class);
		job.setOutputValueClass(IntWritable.class);
		
		// 6 設置輸入和輸出路徑
		FileInputFormat.setInputPaths(job, new Path(args[0]));
		FileOutputFormat.setOutputPath(job, new Path(args[1]));

		// 7 提交job
		boolean result = job.waitForCompletion(true);
		System.exit(result ? 0 : 1);
	}
}

• 可以看到hadoop默認會對數(shù)據(jù)進行排序
• 如果此時再次點擊運行的話，會報錯，顯示輸出路徑存在；因此對于mapreduce程序，如果輸出路徑存在了，就會報錯。

5）本地測試

（1）需要首先配置好HADOOP_HOME變量以及Windows運行依賴

（2）在IDEA/Eclipse上運行程序

2.2 提交到集群測試

集群上測試

（1）用maven打jar包，需要添加的打包插件依賴

<build>
    <plugins>
        <plugin>
            <artifactId>maven-compiler-plugin</artifactId>
            <version>3.6.1</version>
            <configuration>
                <source>1.8</source>
                <target>1.8</target>
            </configuration>
        </plugin>
        <plugin>
            <artifactId>maven-assembly-plugin</artifactId>
            <configuration>
                <descriptorRefs>
                    <descriptorRef>jar-with-dependencies</descriptorRef>
                </descriptorRefs>
            </configuration>
            <executions>
                <execution>
                    <id>make-assembly</id>
                    <phase>package</phase>
                    <goals>
                        <goal>single</goal>
                    </goals>
                </execution>
            </executions>
        </plugin>
    </plugins>
</build>

注意：如果工程上顯示紅叉。在項目上右鍵->maven->Reimport刷新即可。

（2）將程序打成jar包

（3）修改不帶依賴的jar包名稱為wc.jar，并拷貝該jar包到Hadoop集群的/home/wenxin/module/hadoop-3.1.3路徑。

（4）啟動Hadoop集群

[root@hadoop102 hadoop-3.1.3]sbin/start-dfs.sh
[root@hadoop103 hadoop-3.1.3]$ sbin/start-yarn.sh

（5）執(zhí)行WordCount程序

[root@hadoop102 hadoop-3.1.3]$ hadoop jar  wc.jar
 com.wenxin.mapreduce.wordcount.WordCountDriver /user/wenxin/input /user/wenxin/output

文章來源地址http://www.zghlxwxcb.cn/news/detail-757378.html

到了這里，關于【Hadoop_06】MapReduce的概述與wc案例的文章就介紹完了。如果您還想了解更多內容，請在右上角搜索TOY模板網以前的文章或繼續(xù)瀏覽下面的相關文章，希望大家以后多多支持TOY模板網！