MapReduce 配置

修改文件

mv /opt/software/hadoop-3.1.3/etc/hadoop/mapred-site.xml.template /opt/software/hadoop-3.1.3/etc/hadoop/mapred-site.xml

修改 mapred-site.xml，在configuration添加如下

  <!-- 指定mapreduce运行在yarn上 -->
  <property>
    <name>mapreduce.framework.name</name>
    <value>yarn</value>
  </property>

yarn 资源调度框架，实现对资源细粒度封装（cpu、内存、带宽），此外，还可以通过yarn协调多种不同计算框架（MR，Spark）

修改 yarn-site.xml，添加内容如下：

  <!-- 指定YARN的ResourceManager的地址，可以直接写 ip -->
  <property>
    <name>yarn.resourcemanager.hostname</name>
    <value>node113</value>
  </property>

  <!-- Reducer获取数据的方式 -->
  <property>
    <name>yarn.nodemanager.aux-services</name>
    <value>mapreduce_shuffle</value>
  </property>

启动相关进程（ResourceManager、NodeManager）

cd /opt/software/hadoop-3.1.3/sbin
./start-yarn.sh

如果启动的时候出现该错误：
ERROR: but there is no YARN_RESOURCEMANAGER_USER defined. Aborting operation.
ERROR: but there is no YARN_NODEMANAGER_USER defined. Aborting operation.

修改 start-yarn.sh 和 stop-yarn.sh，添加如下

YARN_RESOURCEMANAGER_USER=root
HADOOP_SECURE_DN_USER=yarn
YARN_NODEMANAGER_USER=root

重新启动后，jps 查看进程，有如下两个，启动成功

118583 ResourceManager
118719 NodeManager

MapReduce 开发

概述

MapReduce 在实际开发过程中分为 Map、Reduce、Job三个不同的工作类，Map用于把datanode上的数据进行计算，Reduce 处理Map得到的结果，Job 是指定和设置Map及Reduce的工作类，输入输出类型，以及文件读取和写入的地址等。

编码

接下来写一个求出每个学生的学分平均值，注意事项已在代码中有写，大家注意看注释。

文件内容

tom 69
tom 84
tom 68
jary 89
jary 90
jary 81
jary 35
rose 23
rose 100
rose 230

定义对象

package com.example.demo.mr;

import lombok.Data;
import org.apache.hadoop.io.Writable;

import java.io.DataInput;
import java.io.DataOutput;
import java.io.IOException;

/**
 * @author big uncle
 * @date 2021/5/11 16:04
 * @module
 **/
@Data
public class Flow implements Writable {

    private String name;

    private Integer score;

    /**
     * 序列化
    **/
    @Override
    public void write(DataOutput dataOutput) throws IOException {
        dataOutput.writeUTF(name);
        dataOutput.writeInt(score);
    }

    /**
     * 反序列化,
     * 必须和序列化的顺序一致
    **/
    @Override
    public void readFields(DataInput dataInput) throws IOException {
        this.name = dataInput.readUTF();
        this.score = dataInput.readInt();
    }
}

编写Map逻辑

package com.example.demo.mr;

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 big uncle
 * @date 2021/5/11 10:46
 * 1. job的 MapTask如何处理文件块数据，是由Mapper组件类来决定的，具体实现需要自己写
 * 2. 开发一个Mapper组件的方式是让一个类实现Mapper类
 * 3. Mapper的第1个泛型对应的 MapTask 的输入Key类型，每行行首偏移量
 * 4. Mapper的第2个泛型对应的 MapTask 的输入value类型，每行内容
 * 5. Mapper的第3个泛型对应的 MapTask 的输出Key类型，对应 map() 的 Context参数
 * 6. Mapper的第4个泛型对应的 MapTask 的输出value类型，对应 map() 的 Context参数
 * 7. 常用的类型 LongWritable IntWritable Text NullWritable
 * 8. 重写 map()方法，map方法会接收 输入key和输入value，有一行数据该方法就会被调用一次。Context 用于输出内容
 **/
public class WordCountMapper3 extends Mapper<LongWritable, Text, Text,Flow > {

    @Override
    protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
        // 读出数据
        String line = value.toString();
        // 切割数据
        String[] words = line.split(" ");
        // 给对象赋值
        Flow flow = new Flow();
        flow.setName(words[0]);
        flow.setScore(Integer.valueOf(words[1]));
        // 以名字为key 对象为值
        context.write(new Text(words[0]),flow);
    }
}

编写Reduce逻辑

package com.example.demo.mr;

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

import java.io.IOException;
import java.util.Iterator;

/**
 * @author big uncle
 * @date 2021/5/11 15:10
 * 1. Reducer第1个参数对应 Mapper的输出key，会自动去重
 * 2. Reducer第2个参数对应 Mapper的输出value，会给重复的数据 每个记 1
 * 3. Reducer第2个参数对应 输出 key 类型
 * 4. Reducer第2个参数对应 输出value 类型
 * 5. Reducer不能单独存在，要接收 Mapper的输出
 * 6. Mapper 可以单独存在，当只有Mapper时，最后的输出结果是Mapper的结果
 * 7. 既有 Mapper 又有 Reducer，输出结果为 Reducer，Mapper 会是中间结果
 **/
public class WordCountReducer3 extends Reducer<Text, Flow,Text,IntWritable> {

    @Override
    protected void reduce(Text key, Iterable<Flow> values, Context context) throws IOException, InterruptedException {
        // 求平均分数
        Iterator<Flow> flows = values.iterator();
        int sum = 0,count=0;
        while (flows.hasNext()){
            Flow flow = flows.next();
            sum += flow.getScore();
            count++;
        }
        context.write(key,new IntWritable(sum/count));
    }
}

编写Job

package com.example.demo.mr;

import org.apache.hadoop.conf.Configuration;
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.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

import java.io.IOException;

/**
 * @author big uncle
 * @date 2021/5/11 11:13
 * @module
 **/
public class WordCountDriver {

    public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException {
        Configuration configuration = new Configuration();
        // 获取job对象1
        Job job = Job.getInstance(configuration);
        // 设置job方法入口的驱动类
        job.setJarByClass(WordCountDriver.class);
        // 设置Mapper工作类
        job.setMapperClass(WordCountMapper3.class);
        // 设置Mapper输出的key的类型
        job.setMapOutputKeyClass(Text.class);
        // 输出Mapper输出的value的类型
        job.setMapOutputValueClass(Flow.class);

        // 设置 Reducer 工作的类
        job.setReducerClass(WordCountReducer3.class);
        // 设置Reducer输出key类型
        job.setOutputKeyClass(Text.class);
        // 设置Reducer输出value类型
        job.setOutputValueClass(IntWritable.class);

        // 指定文件存储路径
        FileInputFormat.setInputPaths(job,new Path("hdfs://node113:9000/test1"));
        // 输出到指定目录，要求结果路径事先不存在
        FileOutputFormat.setOutputPath(job,new Path("hdfs://node113:9000/test1/result"));

        // 提交job
        job.waitForCompletion(true);
    }
}

输出结果

jary    73
rose    117
tom 73

自定义分区

package com.example.demo.mr;

import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Partitioner;

/**
 * @author big uncle
 * @date 2021/5/11 16:54
 * 1. 自定义一个分区器
 * 2. 第 1 个泛型是Mapper的输出Key类型
 * 3. 第 2 个泛型是Mapper的输出value类型
 * 4. 分区是有分区编号的 从0开始。如果有三个分区，即 0，1，2
 **/
public class FlowPartitioner extends Partitioner<Text,Flow> {

    @Override
    public int getPartition(Text text, Flow flow, int i) {
        if(text.toString().equals("tom")){
            return 0;
        } else if(text.toString().equals("jary")){
            return 1;
        } else {
            return 2;
        }
    }
}

使用分区，在以上的 Job 编写，添加如下就行

        // 设置 Reducer 启动数量
        job.setNumReduceTasks(2);
        // 设置自定义分区类
        job.setPartitionerClass(FlowPartitioner.class);

合并组件（Combiner）

Combiner 组件依然是 继承 Reducer 类

package com.example.demo.mr;

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

import java.io.IOException;
import java.util.Iterator;

/**
 * @author big uncle
 * @date 2021/5/12 11:18
 * @module
 **/
public class WordCountCombiner extends Reducer<Text, Flow,Text,Flow> {

    /**
     * 提前做部分累加
    **/
    @Override
    protected void reduce(Text key, Iterable<Flow> values, Context context) throws IOException, InterruptedException {
        Iterator<Flow> flows = values.iterator();
        int sum = 0;
        Flow f = new Flow();
        while (flows.hasNext()){
            Flow flow = flows.next();
            if(null == f.getScore()){
                f.setScore(0);
            }
            f.setScore(f.getScore()+flow.getScore());
        }
        context.write(key,f);
    }
}

使用组件，在以上的 Job 编写，添加如下就行

        // 设置 Combiner 组件
        job.setCombinerClass(WordCountCombiner.class);

数据倾斜的发生

数据经过 MapTask后，由于不同key 的数据量分布不均，在 MapTask 阶段中通过 分区 将相同的 key 的数据 spill （溢写）写入磁盘，最后merge成最终map阶段输出文件，交由 ReduceTask处理。如此一来 80G 的 aa 将发往同一个 reducer 。

数据倾斜的解决

用 job 提供的缓存解决，只建议缓存小文件

我们把重要数据提前缓存到 MapTask，尽量先在 MapTask 把我们的逻辑处理掉。
job 中加入如下

job.addCacheFile(new Path("hdfs://ip:port/文件路径.txt").toUri())

MapTask，重写 setup() 初始化方法，用于提前加载缓存文件，封装到Task的全局变量中，直到处理业务时配合使用。

    @Override
    protected void setup(Context context) throws IOException, InterruptedException {
        // 得到job的工作配置
        Configuration configuration = context.getConfiguration();
        // 得到在job 缓存的文件
        URI[] uris = context.getCacheFiles();
        // 得到文件系统
        FileSystem fileSystem = FileSystem.get(uris[0],configuration);
        InputStream inputStream = fileSystem.open(new Path(uris[0]));
        BufferedReader br = new BufferedReader(new InputStreamReader(inputStream));
        String line = null;
        while ((line=br.readLine())!=null){
            String[] items = line.split(",");
            // 处理文件
        }
        br.close();
    }

用修改文件内容的方式

通过在文件内容前添加一定的数字或单词等，比如奇偶数，可以在MapTask中根据奇偶进行分区文件溢写(同样的内容分为多个小文件)，就不会导致文件过大，ReduceTask处理大文件。

将多个切片合为一个

该方式适合多个小文件合并。

// 设置 MapTask的切片合并为一个
job.setInputFormatClass(CombineTextInputFormat.class);
// 可以通过第二个参数设置 MapTask根据文件内容大小合并为几个，默认会完成行的追溯（完整性）
CombineTextInputFormat.setMaxInputSplitSize(job,100);