您的当前位置：首页尚硅谷hadoop3.x-MapReduce（1）

尚硅谷hadoop3.x-MapReduce（1）

来源：图艺博知识网

MapReduce 概述

1. MapReduce优缺点

优点：

缺点：

不擅长实时计算

MapReduce 无法像 MySQL 一样，在毫秒或者秒级内返回结果。
不擅长流式计算

流式计算的输入数据是动态的，而 MapReduce 的输入数据集是静态的，不能动态变化。这是因为 MapReduce 自身的设计特点决定了数据源必须是静态的。
不擅长 DAG（有向无环图）计算

多个应用程序存在依赖关系，后一个应用程序的输入为前一个的输出。在这种情况下， MapReduce 并不是不能做，而是使用后，每个 MapReduce 作业的输出结果都会写入到磁盘，会造成大量的磁盘 IO，导致性能非常的低下。

2. MapReduce 核心思想

分布式的运算程序往往需要分成至少2个阶段：

第一个阶段的MapTask并发实例，完全并行运行，互不相干。
第二个阶段的 ReduceTask 并发实例互不相干，但是他们的数据依赖于上一个阶段的所有 MapTask 并发实例的输出。

MapReduce 编程模型只能包含一个 Map 阶段和一个 Reduce 阶段，如果用户的业务逻辑非常复杂，那就只能多个 MapReduce 程序，串行运行。

3. MapReduce 进程

一个完整的 MapReduce 程序在分布式运行时有三类实例进程：

MrAppMaster：负责整个程序的过程调度及状态协调。
MapTask：负责 Map 阶段的整个数据处理流程。
ReduceTask：负责 Reduce 阶段的整个数据处理流程。

4. 官方 WordCount

采用反编译工具反编译源码，发现 WordCount 案例有 Map 类、Reduce 类和驱动类。且数据的类型是 Hadoop 自身封装的序列化类型。

常用数据序列化类型：

5. MapReduce 编程规范

用户编写的程序分成三个部分：Mapper、Reducer 和 Driver。

6. WordCount 案例实操

本地测试：

需求：统计一堆文件中单词出现的个数（WordCount案例）

环境准备：

在idea新建Maven工程MapReduceDemo后，在pom.xml文件</project>前添加下列内容：

<dependencies>
        <dependency>
            <groupId>org.apache.hadoop</groupId>
            <artifactId>hadoop-client</artifactId>
            <version>3.3.6</version>
        </dependency>
        <dependency>
            <groupId>junit</groupId>
            <artifactId>junit</artifactId>
            <version>4.13.2</version>
        </dependency>
        <dependency>
            <groupId>org.slf4j</groupId>
            <artifactId>slf4j-log4j12</artifactId>
            <version>2.0.5</version>
        </dependency>
    </dependencies>

在项目的 src/main/resources 目录下，新建一个文件，命名为“log4j.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

创建包名com.用户名.mapreduce.wordcount

编写程序：

Mapper类：

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;

public class WordCountMapper extends Mapper<LongWritable, Text, Text, IntWritable> {

    Text k = new Text();
    IntWritable v = new IntWritable(1);

    @Override
    protected void map(LongWritable key, Text value, Mapper<LongWritable, Text, Text, IntWritable>.Context context) throws IOException, InterruptedException {
        // 1. 获取一行
        String line = value.toString();

        // 2. 切割
        String[] words = line.split(" ");

        // 3. 输出
        for (String word : words) {
            k.set(word);
            context.write(k, v);
        }
    }
}

Reducer类：

import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;

import java.io.IOException;

public class WordCountReducer extends Reducer<Text, IntWritable, Text, IntWritable> {

    int sum;
    IntWritable v = new IntWritable();

    @Override
    protected void reduce(Text key, Iterable<IntWritable> values, Reducer<Text, IntWritable, Text, IntWritable>.Context context) throws IOException, InterruptedException {

        // 1. 累加求和
        sum = 0;
        for (IntWritable value : values) {
            sum += value.get();
        }

        // 2. 输出
        v.set(sum);
        context.write(key, v);
    }
}

Driver驱动类：

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;

import java.io.IOException;

public class WordCountDriver {

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

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

        // 2. 关联本Driver程序的jar
        job.setJarByClass(WordCountDriver.class);

        // 3. 关联Mapper和Reducer的jar
        job.setMapperClass(WordCountMapper.class);
        job.setReducerClass(WordCountReducer.class);

        // 4. 设置Mapper输出的key和value类型
        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(IntWritable.class);

        // 5. 设置最终输出的key和value类型
        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);
    }
}

提交到集群测试

用 maven 打 jar 包，在pom.xml文件添加下列打包插件依赖：

<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>

注：有三处地方会爆红，但是可以不用管，可以正常运行，我也不知道为什么。

将程序打成 jar 包

成功后会出现两个jar包，一个是不带依赖的，很小，另一个带了依赖，大很多。但是我们只需要将不带依赖的上传到集群。
修改不带依赖的 jar 包名称为 wc.jar，并拷贝该 jar 包到 Hadoop 集群的 /opt/module/hadoop-3.3.6 路径。
启动集群并执行WordCount程序

hadoop jar  wc.jar Driver类的全类名 /input /output

注意：此处在wc.jar后面需要跟上Driver类的全类名，因为我们的程序中Driver类有main方法，是程序的入口。

只需在Driver类页面右击鼠标，光标移到Copy/Paste Special，选中Copy Reference即可。

因篇幅问题不能全部显示，请点此查看更多更全内容

查看全文