MapReduce高阶训练及Yarn资源调度

为什么JavaBean要继承Writable和WritableComparable接口

为什么javaBean要继承Writable和WritableComparable接口？
1. 如果一个javaBean想要作为MapReduce的key或者value，就一定要实现序列化，因为
2. 在Map到Reduce阶段的时候，只能是传输二进制数据，不可能将字符流直接进行RPC传输，
只要一个javabean实现了序列化和反序列化，就可以做为key或者value

最简单的序列化和反序列化就是实现Writable接口

ps：javaBean在作为key的时候有点不同，除了要继承Writable接口还需要
实现Comparable接口

因为在shuffle到Reduce阶段的合并阶段，需要根据key对数据进行排序，合并，如果
不实现这个接口，运行时会出错

WritableComparable就是Writable接口和java.lang.Comparable<T>的一个子接口，
所以将要作为key的javaBean直接继承WritableComparable就可以了

2. java序列化与Writable序列化的比较

2.1 java序列化不够灵活，为了更好的控制序列化的整个流程所以使用Writable

2.2 java序列化不符合序列化的标准，没有做一定的压缩，java序列化首先写类名，
然后再是整个类的数据，而且成员对象在序列化中只存引用，成员对象的可以出现的位置很
随机，既可以在序列化的对象前，也可以在其后面，这样就对随机访问造成影响，一旦出错，
整个后面的序列化就会全部错误

2.3 Java序列化每次序列化都要重新创建对象，内存消耗大，而Writable是可以重用的

一、社交粉丝数据分析

用户：用户的好友列表
A:B,C,D,F,E,O
B:A,C,E,K
C:F,A,D,I
D:A,E,F,L
E:B,C,D,M,L
F:A,B,C,D,E,O,M
G:A,C,D,E,F
H:A,C,D,E,O
I:A,O
J:B,O
K:A,C,D
L:D,E,F
M:E,F,G
O:A,H,I,J
求：哪两个人之间有共同好友，以及他们的共同好友都有哪些人

思路分析：

C这个好友，出现在哪些用户列表里面
*1.第一个mapreduce实现这样的结果
第一步：读取数据
A:B,C,D,F,E,O
按照冒号进行切分
然后将好友列表继续按照逗号切分
String[]  arrays = ["B,C,D,F,E,O"]
String[]  arrays2 = E：B,C,D,M,L
map阶段发出去的数据
key2    value2
B 		A
C 		A
D 		A
F 		A
E 		A
O 		A
B 		E

数据到了reduce
B [A ，E]

有一个共同好友是B
A-E   B

数据输出成这样的形式
A-B-E-F-G-H-K  C
A-B            E

*2.第二个mapreduce
用户列表           好友
 F-D-O-I-H-B-K-G-C-	A
 E-A-J-F-	B
 K-A-B-E-F-G-H-	C
 G-K-C-A-E-L-F-H-	D
 G-F-M-B-H-A-L-D-	E
 M-D-L-A-C-G-	F
 M-	G
 O-	H
 C-O-	I
 O-	J
 B-	K
 E-D-	L
 F-E-	M
 J-I-H-A-F-	O
 
 F-D   F-O  F----C
 G-C
 
 避免出现 A-E B
          A-E D
		  
		  A-E  B,D
		  A-E   B-D

到reduce阶段
A-B  [C,E]

A-B  C-E

step1-mapreduce

package cn.itcast.mr.demo6.step1;

import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;

import java.io.IOException;

public class Step1Mapper extends Mapper<LongWritable, Text,Text,Text> {
    @Override
    protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
        //用户与好友列表
        String[] split = value.toString().split(":");
        //好友列表
        String[] split1 = split[1].split(",");
        for (String friend : split1) {
            context.write(new Text(friend),new Text(split[0]));
        }
    }
}

package cn.itcast.mr.demo6.step1;

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

import java.io.IOException;

/*最终输出结果
用户列表            好友
F-D-O-I-H-B-K-G-C-	A
E-A-J-F-	B
K-A-B-E-F-G-H-	C
G-K-C-A-E-L-F-H-	D
G-F-M-B-H-A-L-D-	E
M-D-L-A-C-G-	F
M-	G
O-	H
C-O-	I
O-	J
B-	K
E-D-	L
F-E-	M
J-I-H-A-F-	O
 */

public class Step1Reducer extends Reducer<Text,Text,Text,Text> {
    //reduce接收到的数据如： B [A,E]
    // B是我们的好友  集合里面装的是多个用户
    //将数据最终转化成 A-B-E-F-G-H-K C
    @Override
    protected void reduce(Text key, Iterable<Text> values, Context context) throws IOException, InterruptedException {
        StringBuilder stringBuilder = new StringBuilder();
        for (Text value : values) {
            stringBuilder.append(value.toString()).append("-");
        }
        context.write(new Text(stringBuilder.toString()),key);
    }
}

package cn.itcast.mr.demo6.step1;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;
import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;

public class Step1Main extends Configured implements Tool {
    @Override
    public int run(String[] args) throws Exception {
        Job job = Job.getInstance(super.getConf(), "step1");
        job.setInputFormatClass(TextInputFormat.class);
        TextInputFormat.addInputPath(job,new Path("file:///xxx//input"));
        //step2
        job.setMapperClass(Step1Mapper.class);
        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(Text.class);

        //step7
        job.setReducerClass(Step1Reducer.class);
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(Text.class);
        //step8
        job.setOutputFormatClass(TextOutputFormat.class);
        TextOutputFormat.setOutputPath(job,new Path("file:///xxx//step1_output"));
        //提交任务
        boolean b = job.waitForCompletion(true);
        return b?0:1;
    }

    public static void main(String[] args) throws Exception {
        int run = ToolRunner.run(new Configuration(), new Step1Main(), args);
        System.exit(run);
    }
}

step2-mapreduce

package cn.itcast.mr.demo6.step2;

import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;

import java.io.IOException;
import java.util.Arrays;

public class Step2Mapper extends Mapper<LongWritable, Text,Text,Text> {
    @Override
    protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
        // F-E-	    M
        //用户列表  好友
        String[] split = value.toString().split("/t");
        //将用户列表拿出来做循环
        String[] userList = split[0].split("-");
        //对我们的用户列表进行排序，避免出现A-E E-A 这种情况
        Arrays.sort(userList);
        for (int i=0;i<userList.length-1;i++){
            for (int j=i+1;j<userList.length;j++){
                String userTwo = userList[i] + "-" + userList[j];
                context.write(new Text(userTwo), new Text(split[1]));
            }
        }
    }
}

package cn.itcast.mr.demo6.step2;

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

import java.io.IOException;

public class Step2Reducer extends Reducer<Text,Text,Text,Text> {
    //A-E B-C-D
    @Override
    protected void reduce(Text key, Iterable<Text> values, Context context) throws IOException, InterruptedException {
        StringBuilder sb = new StringBuilder();
        for (Text value : values) {
            sb.append(value.toString()).append("-");
        }
        context.write(key,new Text(sb.toString()));
    }
}

package cn.itcast.mr.demo6.step2;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;
import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;

public class Step2Main extends Configured implements Tool {
    @Override
    public int run(String[] args) throws Exception {
        Job job = Job.getInstance(super.getConf(), "step2");
        job.setInputFormatClass(TextInputFormat.class);
        TextInputFormat.addInputPath(job, new Path("file:///xxx//step1_output"));

        job.setMapperClass(Step2Mapper.class);
        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(Text.class);

        job.setReducerClass(Step2Reducer.class);
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(Text.class);

        job.setOutputFormatClass(TextOutputFormat.class);
        TextOutputFormat.setOutputPath(job,new Path("file:///xxx//step2_output"));

        boolean b = job.waitForCompletion(true);

        return b?0:1;
    }

    public static void main(String[] args) throws Exception {
        int run = ToolRunner.run(new Configuration(), new Step2Main(), args);
        System.exit(run);
    }
}

二、倒排索引建立

需求分析

倒排索引：求取哪些单词在哪些文章里面出现过多少次
a.txt
hello
world
hello


b.txt  
hello
abc

hello  a.txt  2
hello  b.txt  1

代码实现

package cn.itcast.mr.demo7;

import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.InputSplit;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.lib.input.FileSplit;

import java.io.IOException;

public class IndexMapper extends Mapper<LongWritable, Text,Text, IntWritable> {
    @Override
    protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
        //获取文章名称
        FileSplit fileSplit = (FileSplit) context.getInputSplit();//方法太少，强转filesplit
        String name = fileSplit.getPath().getName();
        String[] s = value.toString().split(" ");
        for (String s1 : s) {
            context.write(new Text(s1+"-"+name),new IntWritable(1));
        }
    }
}

package cn.itcast.mr.demo7;

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

import java.io.IOException;

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

package cn.itcast.mr.demo7;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.conf.Configured;
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.TextInputFormat;
import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;

public class IndexMain extends Configured implements Tool {
    @Override
    public int run(String[] args) throws Exception {
        Job job = Job.getInstance(super.getConf(), "index");
        job.setInputFormatClass(TextInputFormat.class);
        TextInputFormat.addInputPath(job, new Path("file:///E://大数据资料//大数据离线资料//5、大数据离线第五天//倒排索引//input"));

        job.setMapperClass(IndexMapper.class);
        job.setMapOutputKeyClass(Text.class);
        job.setOutputValueClass(IntWritable.class);

        job.setReducerClass(IndexReducer.class);
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(IntWritable.class);

        job.setOutputFormatClass(TextOutputFormat.class);
        TextOutputFormat.setOutputPath(job,new Path("file:///E://大数据资料//大数据离线资料//5、大数据离线第五天//倒排索引//output"));

        boolean b = job.waitForCompletion(true);

        return b?0:1;
    }

    public static void main(String[] args) throws Exception {
        int run = ToolRunner.run(new Configuration(), new IndexMain(), args);
        System.exit(run);
    }
}

三、自定义inputForm合并小文件

1.1 需求

无论hdfs还是mapreduce，对于小文件都有损效率，实践中，又难免面临处理大量小文件的场景，此时，就需要有相应解决方案

1.2 分析

1.3 实现

MyInputFormat

package cn.itcast.mr.demo8;

import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.BytesWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.mapreduce.InputSplit;
import org.apache.hadoop.mapreduce.JobContext;
import org.apache.hadoop.mapreduce.RecordReader;
import org.apache.hadoop.mapreduce.TaskAttemptContext;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;

import java.io.IOException;

public class MyInputFormat extends FileInputFormat<NullWritable, BytesWritable> {
    /*
    这个方法返回值是一个RecordReader
     */
    @Override
    public RecordReader<NullWritable, BytesWritable> createRecordReader(InputSplit split, TaskAttemptContext context) throws IOException, InterruptedException {
        MyRecordReader myRecordReader = new MyRecordReader();
        myRecordReader.initialize(split,context);
        return myRecordReader;
    }

    /*
    表示文件是否可切分，直接返回false，表示文件不可切分
    目的：读文件的时候，就会一次性将文件内容全部读取出来
     */
    @Override
    protected boolean isSplitable(JobContext context, Path filename) {
        return false;
    }
}

MyRecordReader

package cn.itcast.mr.demo8;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FSDataInputStream;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.BytesWritable;
import org.apache.hadoop.io.IOUtils;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.InputSplit;
import org.apache.hadoop.mapreduce.RecordReader;
import org.apache.hadoop.mapreduce.TaskAttemptContext;
import org.apache.hadoop.mapreduce.lib.input.FileSplit;

import java.io.IOException;

public class MyRecordReader extends RecordReader<NullWritable, BytesWritable> {
    private FileSplit fileSplit;
    private Configuration configuration;
    //这里定义我们的v2
    private BytesWritable bytesWritable = new BytesWritable();
    private boolean nextKeyValue = false;

    /**
     * 初始化方法
     * @param split 文件切片：拿到文件切片就可以拿到对应文件，拿到文件就可以将文件转换成字节数组
     * @param context 上下文对象，我们的参数都封装在context
     * @throws IOException
     * @throws InterruptedException
     */
    @Override
    public void initialize(InputSplit split, TaskAttemptContext context) throws IOException, InterruptedException {
        this.fileSplit = (FileSplit) split;
        this.configuration = context.getConfiguration();

    }

    /**
     * 往下继续读取文件
     * 返回值是boolean,如果返回false，表示继续往下读取
     * 如果返回true，表示文件已经完成读取，不需要再继续往下读取了
     * @return
     * @throws IOException
     * @throws InterruptedException
     */
    @Override
    public boolean nextKeyValue() throws IOException, InterruptedException {
        if (!nextKeyValue){
            //我们需要将文件内容读取出来，封装到BytesWritable里面进行返回
            //字节数组定义的大小，需要装得下我们的文件内容
            byte[] fileContent = new byte[(int)fileSplit.getLength()];
            Path path = fileSplit.getPath();
            //file:///  or hdfs://
            //获取文件系统
            FileSystem fileSystem = path.getFileSystem(configuration);
            //拿到文件系统，我们就可以打开文件输入流
            FSDataInputStream inputStream = fileSystem.open(path);
            //如何将输入流转换到字节数组里面去
            IOUtils.readFully(inputStream,fileContent,0,(int)fileSplit.getLength());
            //bytesWritable.set(fileContent,0,(int)fileSplit.getLength());
            bytesWritable.set(fileContent,0,fileContent.length);
            //将我们读取文件的标识置为true，表示我们的文件已经读取完成
            nextKeyValue=true;
            IOUtils.closeStream(inputStream);
            return nextKeyValue;
        }
        return false;
    }

    /**
     * 这个方法返回我们的k1
     * @return
     * @throws IOException
     * @throws InterruptedException
     */
    @Override
    public NullWritable getCurrentKey() throws IOException, InterruptedException {
        return NullWritable.get();
    }

    /**
     * 这个方法返回我们的v1 BytesWritable    我们需要将文件内容读取出来，封装到BytesWritable里面进行返回
     * @return
     * @throws IOException
     * @throws InterruptedException
     */
    @Override
    public BytesWritable getCurrentValue() throws IOException, InterruptedException {
        return bytesWritable;
    }

    /**
     * 文件读取进度的方法，没什么用
     * @return
     * @throws IOException
     * @throws InterruptedException
     */
    @Override
    public float getProgress() throws IOException, InterruptedException {
        return nextKeyValue?1.0f:0.0f;
    }

    /**
     * 读取完成释放资源
     * @throws IOException
     */
    @Override
    public void close() throws IOException {
    }
}

MyMapperInput

package cn.itcast.mr.demo8;

import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.BytesWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.InputSplit;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.lib.input.FileSplit;

import java.io.IOException;

public class MyMapperInput extends Mapper<NullWritable, BytesWritable, Text,BytesWritable> {
    @Override
    protected void map(NullWritable key, BytesWritable value, Context context) throws IOException, InterruptedException {
        //获取文件切片
        FileSplit inputSplit = (FileSplit)context.getInputSplit();
        /*Path path = inputSplit.getPath();
        String s = path.toString();*/
        String name = inputSplit.getPath().getName();
        context.write(new Text(name),value);
    }
}

OwnInputFormatMain

package cn.itcast.mr.demo8;

import jdk.nashorn.internal.scripts.JO;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.BytesWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.output.SequenceFileOutputFormat;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;

public class OwnInputFormatMain extends Configured implements Tool {
    @Override
    public int run(String[] args) throws Exception {
        Job job = Job.getInstance(super.getConf(), "inputformat");
        job.setInputFormatClass(MyInputFormat.class);
        MyInputFormat.addInputPath(job, new Path("file:///E://大数据资料//大数据离线资料//5、大数据离线第五天//自定义inputformat_小文件合并//input"));

        job.setMapperClass(MyMapperInput.class);
        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(BytesWritable.class);

        //虽然没有reduce，但是默认输出类型是LongWritable Text
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(BytesWritable.class);

        job.setOutputFormatClass(SequenceFileOutputFormat.class);
        SequenceFileOutputFormat.setOutputPath(job, new Path("file:///E://大数据资料//大数据离线资料//5、大数据离线第五天//自定义inputformat_小文件合并//output_Sequence"));

        boolean b = job.waitForCompletion(true);

        return b?0:1;
    }

    public static void main(String[] args) throws Exception {
        int run = ToolRunner.run(new Configuration(), new OwnInputFormatMain(), args);
        System.exit(run);
    }
}

四、自定义outputFormat

2.1 需求

现在有一些订单的评论数据，需求，将订单的好评与差评进行区分开来，将最终的数据分开到不同的文件夹下面去，数据内容参见资料文件夹，其中数据第九个字段表示好评，中评，差评。0：好评，1：中评，2：差评

2.2 分析

2.3 实现

MyOutputFormat

package cn.itcast.mr.demo9;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FSDataOutputStream;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.RecordWriter;
import org.apache.hadoop.mapreduce.TaskAttemptContext;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

import java.io.IOException;

public class MyOutputFormat extends FileOutputFormat<Text, NullWritable> {
    @Override
    public RecordWriter<Text, NullWritable> getRecordWriter(TaskAttemptContext context) throws IOException, InterruptedException {
        //从这个方法中可以获得一个configuration
        Configuration configuration = context.getConfiguration();
        FileSystem fileSystem = FileSystem.get(configuration);
        //好评文件输出路径
        Path goodComment = new Path("file:///xxx.txt");
        //差评文件输出路径
        Path badComment = new Path("file:///xxxx.txt");
        //获取两个文件输出流
        FSDataOutputStream fsDataOutputStream = fileSystem.create(goodComment);
        FSDataOutputStream fsDataOutputStream1 = fileSystem.create(badComment);

        MyRecordWriter myRecordWriter = new MyRecordWriter(fsDataOutputStream, fsDataOutputStream1);

        return myRecordWriter;
    }
}

MyRecordWriter

package cn.itcast.mr.demo9;

import org.apache.hadoop.fs.FSDataOutputStream;
import org.apache.hadoop.io.IOUtils;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.RecordWriter;
import org.apache.hadoop.mapreduce.TaskAttemptContext;

import java.io.IOException;

public class MyRecordWriter extends RecordWriter<Text, NullWritable> {
    private FSDataOutputStream goodStream;
    private FSDataOutputStream badStream;

    public MyRecordWriter(FSDataOutputStream goodStream,FSDataOutputStream badStream){
        this.goodStream = goodStream;
        this.badStream = badStream;
    }

    public MyRecordWriter(){}

    /**
     * 这个write方法就是往外些数据，我们可以根据这个key，来判断文件究竟往哪个目录下面写
     * @param key
     * @param nullWritable
     * @throws IOException
     * @throws InterruptedException
     */
    @Override
    public void write(Text key, NullWritable nullWritable) throws IOException, InterruptedException {
        String[] split = key.toString().split("/t");
        // 0 好评 ；1 中评；2 差评
        if (Integer.parseInt(split[9])<=1){
            //好评
            goodStream.write(key.getBytes());
            goodStream.write("/r/n".getBytes());
        }
        else {
            //差评
            badStream.write(key.getBytes());
            badStream.write("/r/n".getBytes());
        }
    }

    @Override
    public void close(TaskAttemptContext taskAttemptContext) throws IOException, InterruptedException {
        IOUtils.closeStream(badStream);
        IOUtils.closeStream(goodStream);
    }
}

MyOutputMapper

package cn.itcast.mr.demo9;

import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;

import java.io.IOException;

public class MyOutputMapper extends Mapper<LongWritable, Text,Text, NullWritable> {
    @Override
    protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
        context.write(value,NullWritable.get());
    }
}

MyOutputMain

package cn.itcast.mr.demo9;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;

public class MyOutputMain extends Configured implements Tool {
    @Override
    public int run(String[] args) throws Exception {
        Job job = Job.getInstance(super.getConf(), "outputformat");
        job.setInputFormatClass(TextInputFormat.class);
        TextInputFormat.addInputPath(job,new Path("file:///xxx//input"));

        job.setMapperClass(MyOutputMapper.class);
        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(NullWritable.class);

        job.setOutputFormatClass(MyOutputFormat.class);
        MyOutputFormat.setOutputPath(job, new Path("file:///xxx//input"));

        boolean b = job.waitForCompletion(true);

        return b?0:1;
    }

    public static void main(String[] args) throws Exception {
        int run = ToolRunner.run(new Configuration(), new MyOutputMain(), args);
        System.exit(run);
    }
}

五、自定义GroupingComparator求取topN

GroupingComparator是mapreduce当中reduce端的一个功能组件，主要的作用是决定哪些数据作为一组，调用一次reduce的逻辑，默认是每个不同的key，作为多个不同的组，每个组调用一次reduce逻辑，我们可以自定义GroupingComparator实现不同的key作为同一个组，调用一次reduce逻辑

3.1 需求

3.2 分析

top2表示的是每个订单当中最大的前两个值
分组求topN
Order_0000001	Pdt_01	222.8
Order_0000001	Pdt_05	25.8

Order_0000002	Pdt_05	722.4
Order_0000002	Pdt_03	522.8

Order_0000003	Pdt_01	222.8

求前top2,出来了两条数据，不满足我们每个组当中取前两个最大的值
select max(price)
from  order o group by o.orderid  order   by  price  desc limit 2

如何通过mapreduce来实现

如何通过MR实现分组求top1
第一个问题：求金额的最大值，要排序，按照金额来进行排序
第二个问题：如何区分不同的订单，，通过订单id来区分我们不同的订单，相同的订单，发送到同一个reduce里面去，形成一个集合

思路：将我们的订单号以及金额同时作为k2,封装到一个JavaBean里面去
	 重写分区规则，按照订单号进行分区
	 重写排序规则，按照金额进行排序

3.3 实现（top1）

第一步：定义OrderBean

package cn.itcast.mr.demo10;

import org.apache.hadoop.io.WritableComparable;

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

public class OrderBean  implements WritableComparable<OrderBean> {

    private String orderId;
    private Double price;

    /*
    安装价格进行排序
     */
    @Override
    public int compareTo(OrderBean o) {
        //需要先比较我们的订单id，如果订单id相同，我们再按照金额进行排序
        //如果订单id不相同，没有可比性
        int result = this.orderId.compareTo(o.orderId);
        if (result == 0){
            //如果订单相同，继续比较价格进行排序
            int i = this.price.compareTo(o.price);
            return -result;
        }
        return result;
    }

    @Override
    public void write(DataOutput out) throws IOException {
        out.writeUTF(orderId);
        out.writeDouble(price);
    }

    @Override
    public void readFields(DataInput in) throws IOException {
        this.orderId = in.readUTF();
        this.price = in.readDouble();
    }

    public String getOrderId() {
        return orderId;
    }

    public void setOrderId(String orderId) {
        this.orderId = orderId;
    }

    public Double getPrice() {
        return price;
    }

    public void setPrice(Double price) {
        this.price = price;
    }

    @Override
    public String toString() {
        return orderId + "/t" + price;
    }
}

第二步：mapper类

package cn.itcast.mr.demo10;

import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;

import java.io.IOException;

public class GroupMapper extends Mapper<LongWritable, Text,OrderBean, NullWritable> {
    @Override
    protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
        String[] split = value.toString().split("/t");
        OrderBean orderBean = new OrderBean();
        orderBean.setOrderId(split[0]);
        orderBean.setPrice(Double.valueOf(split[2]));
        context.write(orderBean,NullWritable.get());
    }
}

第三步：reducer类

package cn.itcast.mr.demo10;

import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;

import java.io.IOException;

public class GroupReducer extends Reducer<OrderBean, NullWritable,OrderBean,NullWritable> {
    @Override
    protected void reduce(OrderBean key, Iterable<NullWritable> values, Context context) throws IOException, InterruptedException {
        context.write(key,NullWritable.get());
    }
}

第四步：自定义分区

package cn.itcast.mr.demo10;

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

public class GroupPartition extends Partitioner<OrderBean, NullWritable> {
    @Override
    public int getPartition(OrderBean orderBean, NullWritable nullWritable, int numPartitions) {
        return (orderBean.getOrderId().hashCode() & Integer.MAX_VALUE) % numPartitions;
    }
}

第五步：自定义groupingComparator

package cn.itcast.mr.demo10;

import org.apache.hadoop.io.WritableComparable;
import org.apache.hadoop.io.WritableComparator;

public class GroupComparator extends WritableComparator {
    //重写默认的无参构造器，然后定义我们反射出来的对象是OrderBean这个类
    public GroupComparator(){
        super(OrderBean.class,true);
    }

    /*
    注意调用compare方法时，参数必须是WritableComparable，因为Object中没有实现compareTo排序
     */
    @Override
    public int compare(WritableComparable a, WritableComparable b) {
        OrderBean first = (OrderBean) a;
        OrderBean second = (OrderBean) b;
        //比较我们的订单id，如果相同的订单id，认为是同一组数据
        return first.getOrderId().compareTo(second.getOrderId());
    }

    /*@Override
    public int compare(Object a, Object b) {
        OrderBean first = (OrderBean) a;
        OrderBean second = (OrderBean) a;
        //比较我们的订单id，如果相同的订单id，认为是同一组数据
        return first.getOrderId().compareTo(second.getOrderId());
    }*/
}

第六步：程序main函数入口

package cn.itcast.mr.demo10;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;
import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;

public class GroupMain extends Configured implements Tool {
    @Override
    public int run(String[] args) throws Exception {
        Job job = Job.getInstance(super.getConf(), "group");
        job.setInputFormatClass(TextInputFormat.class);
        TextInputFormat.addInputPath(job,new Path("file:///E://大数据资料//大数据离线资料//5、大数据离线第五天//自定义groupingComparator//input"));

        job.setMapperClass(GroupMapper.class);
        job.setMapOutputKeyClass(OrderBean.class);
        job.setMapOutputValueClass(NullWritable.class);

        job.setPartitionerClass(GroupPartition.class);

        job.setGroupingComparatorClass(GroupComparator.class);

        job.setReducerClass(GroupReducer.class);
        job.setOutputKeyClass(OrderBean.class);
        job.setOutputValueClass(NullWritable.class);

        job.setOutputFormatClass(TextOutputFormat.class);
        TextOutputFormat.setOutputPath(job,new Path("file:///E://大数据资料//大数据离线资料//5、大数据离线第五天//自定义groupingComparator//output_top1"));

        boolean b = job.waitForCompletion(true);

        return b?0:1;
    }

    public static void main(String[] args) throws Exception {
        int run = ToolRunner.run(new Configuration(), new GroupMain(), args);
        System.exit(run);
    }
}

3.3 实现（topN）

在原来top1改进topN

第一步：mapper类

public class GroupMapper extends Mapper<LongWritable, Text,OrderBean, DoubleWritable> {
    @Override
    protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
        String[] split = value.toString().split("/t");
        OrderBean orderBean = new OrderBean();
        orderBean.setOrderId(split[0]);
        orderBean.setPrice(Double.valueOf(split[2]));
        context.write(orderBean,new DoubleWritable(Double.valueOf(split[2])));
    }
}

第二部：reducer类

public class GroupReducer extends Reducer<OrderBean, DoubleWritable,OrderBean,DoubleWritable> {
    @Override
    protected void reduce(OrderBean key, Iterable<DoubleWritable> values, Context context) throws IOException, InterruptedException {
        int i=0;
        for (DoubleWritable value : values) {
            i++;
            if (i<=2){
                context.write(key,value);
            }else {
                break;
            }
        }
    }
}

第三步：自定义分区

public class GroupPartition extends Partitioner<OrderBean, DoubleWritable> {
    @Override
    public int getPartition(OrderBean orderBean, DoubleWritable doubleWritable, int numPartitions) {
        return (orderBean.getOrderId().hashCode() & Integer.MAX_VALUE) % numPartitions;
    }
}

第四步：main函数

job.setMapperClass(GroupMapper.class);
job.setMapOutputKeyClass(OrderBean.class);
job.setMapOutputValueClass(DoubleWritable.class);

job.setPartitionerClass(GroupPartition.class);

job.setGroupingComparatorClass(GroupComparator.class);

job.setReducerClass(GroupReducer.class);
job.setOutputKeyClass(OrderBean.class);
job.setOutputValueClass(DoubleWritable.class);

六、分区与分组的关系：

分区与分组的关系：
分区：主要是决定了我们的数据去往哪一个reduce
分组：决定了哪些数据分到一个组里面去，形成一个集合，一起调用一次reduce逻辑

七、Mapreduce的其他补充

多job串联
一个稍复杂点的处理逻辑往往需要多个mapreduce程序串联处理，多job的串联可以借助mapreduce框架的JobControl实现

ControlledJob cJob1 = new ControlledJob(job1.getConfiguration());
ControlledJob cJob2 = new ControlledJob(job2.getConfiguration());
ControlledJob cJob3 = new ControlledJob(job3.getConfiguration());
cJob1.setJob(job1);
cJob2.setJob(job2);
cJob3.setJob(job3);
// 设置作业依赖关系
cJob2.addDependingJob(cJob1);
cJob3.addDependingJob(cJob2);
JobControl jobControl = new JobControl("RecommendationJob");
jobControl.addJob(cJob1);
jobControl.addJob(cJob2);
jobControl.addJob(cJob3);
// 新建一个线程来运行已加入JobControl中的作业，开始进程并等待结束
Thread jobControlThread = new Thread(jobControl);
jobControlThread.start();
while (!jobControl.allFinished()) {
    Thread.sleep(500);
}
jobControl.stop();

return 0;

八、mapreduce参数优化

mapreduce任务提交的过程

1 资源相关参数

以下调整参数都在mapred-site.xml这个配置文件当中有

//以下参数是在用户自己的mr应用程序中配置就可以生效
(1) mapreduce.map.memory.mb: 一个Map Task可使用的资源上限（单位:MB），
默认为1024。如果Map Task实际使用的资源量超过该值，则会被强制杀死。
(2) mapreduce.reduce.memory.mb: 一个Reduce Task可使用的资源上限（单位:MB），
默认为1024。如果Reduce Task实际使用的资源量超过该值，则会被强制杀死。
(3) mapred.child.java.opts  配置每个map或者reduce使用的内存的大小，
默认是200M
(4) mapreduce.map.cpu.vcores: 每个Map task可使用的最多cpu core数目,
 默认值: 1
(5) mapreduce.reduce.cpu.vcores: 每个Reduce task可使用的最多cpu core数目, 
默认值: 1
virtual 虚拟的

//shuffle性能优化的关键参数，应在yarn启动之前就配置好
(6)mapreduce.task.io.sort.mb   100         //shuffle的环形缓冲区大小，默认100m
(7)mapreduce.map.sort.spill.percent   0.8    //环形缓冲区溢出的阈值，默认80%

//应该在yarn启动之前就配置在服务器的配置文件中才能生效
以下配置都在yarn-site.xml配置文件当中配置
(8) yarn.scheduler.minimum-allocation-mb	  1024   给应用程序container分配的最小内存
(9) yarn.scheduler.maximum-allocation-mb	  8192	给应用程序container分配的最大内存
(10) yarn.scheduler.minimum-allocation-vcores	1	container最小的虚拟内核的个数
(11)yarn.scheduler.maximum-allocation-vcores	32 container最大的虚拟内核的个数
(12)yarn.nodemanager.resource.memory-mb   8192  每个nodemanager给多少内存

2 容错相关参数

(1) mapreduce.map.maxattempts: 每个Map Task最大重试次数，一旦重试参数超过该值，
则认为Map Task运行失败，默认值：4。
(2) mapreduce.reduce.maxattempts: 每个Reduce Task最大重试次数，一旦重试参数
超过该值，则认为Map Task运行失败，默认值：4。
(3) mapreduce.job.maxtaskfailures.per.tracker: 当失败的Map Task失败比例
超过该值为，整个作业则失败，默认值为0. 如果你的应用程序允许丢弃部分输入数据，则
该值设为一个大于0的值，比如5，表示如果有低于5%的Map Task失败
（如果一个Map Task重试次数超过mapreduce.map.maxattempts，则认为这个Map Task
失败，其对应的输入数据将不会产生任何结果），整个作业仍认为成功。
(5) mapreduce.task.timeout: Task超时时间，默认值为600000毫秒，经常需要设置的
一个参数，该参数表达的意思为：如果一个task在一定时间内没有任何进入，即不会读取新的
数据，也没有输出数据，则认为该task处于block状态，可能是卡住了，也许永远会卡主，
为了防止因为用户程序永远block住不退出，则强制设置了一个该超时时间（单位毫秒）。
如果你的程序对每条输入数据的处理时间过长（比如会访问数据库，通过网络拉取数据等），
建议将该参数调大，该参数过小常出现的错误提示是
“AttemptID:attempt_14267829456721_123456_m_000224_0 Timed out after 300 secsContainer killed by the ApplicationMaster.”。

3 效率和稳定性相关参数

数据倾斜出现，推测执行机制解决不了这个问题，只会造成资源更紧张，建议关闭

(1) mapreduce.map.speculative: 是否为Map Task打开推测执行机制，默认为true，
如果为true，如果Map执行时间比较长，那么集群就会推测这个Map已经卡住了，会重新启动
同样的Map进行并行的执行，哪个先执行完了，就采取哪个的结果来作为最终结果，一般直接
关闭推测执行
(2) mapreduce.reduce.speculative: 是否为Reduce Task打开推测执行机制，默认
为true，如果reduce执行时间比较长，那么集群就会推测这个reduce已经卡住了，会重新
启动同样的reduce进行并行的执行，哪个先执行完了，就采取哪个的结果来作为最终结果，
一般直接关闭推测执行
(3) mapreduce.input.fileinputformat.split.minsize: FileInputFormat做
切片时的最小切片大小，默认为0
(4) mapreduce.input.fileinputformat.split.maxsize:  FileInputFormat做
切片时的最大切片大小
（已过时的配置，2.7.5当中直接把这个配置写死了，写成了Integer.maxValue的值）
(切片的默认大小就等于blocksize，即 134217728)

九、Yarn资源调度管理

yarn的资源调度管理：

yarn是我们hadoop2.x当中引进的一个新的模块，主要用于管理我们集群当中的资源
比如说：内存，cpu
yarn不光管理硬件资源，还管理运行的一些任务信息等等
yarn的调度可以分为两个层级来说
一级管理调度：
	管理计算机的资源（CPU,内存，网络IO，磁盘）
	运行的job任务的生命周（每一个应用执行的情况，都需要汇报给ResourceManager）
二级管理调度：
	任务的计算模型
	多样化的计算模型 storm spark
	
yarn的官网文档说明：
http://hadoop.apache.org/docs/r2.7.5/hadoop-yarn/hadoop-yarn-site/YARN.html

Yarn的主要组件介绍与作用

yarn当中各个主要组件的作用：

ResourceManager：yarn集群的主节点，主要用于接收客户端提交的任务，并对资源进行分配

NodeManager：yarn集群的从节点，主要用于任务的计算

ApplicationMaster：当有新的任务提交到ResourceManager的时候，ResourceManager
会在某个从节点nodeManager上面启动一个ApplicationMaster进程
主要职责：申请资源
        分配资源（分配container）
		监控任务执行的进度状况
		回收资源
		与resourceManager通信，报告任务的执行状况

Container:资源分配的单位，所有的资源都是以container的形式来进行划分的，
便于资源的分配和回收

jobHistory：历史完成的任务的日志信息

TimeLineServer:  2.4版本以后出来的新特性，查看正在执行的任务的信息

yarn的发展历程以及详细介绍：
https://www.ibm.com/developerworks/cn/opensource/os-cn-hadoop-yarn/

yarn当中的调度器

调度器主要解决的是任务先后提交，如何保证任务最快执行的一种策略
研究的是任务之间如何一起执行的问题

第一种调度器：FIFO Scheduler （队列调度器）

1：fifo  队列调度器，first  in  first  out 没人用  
第一个任务来了，先执行，第二个任务来了，等着
如果一个很大的计算任务先来，需要执行两个小时，再来一个小任务，需要执行两分钟

第二种调度器：capacity scheduler（容量调度器，apache版本默认使用的调度器）

容量调度器：将我们集群的资源，划分成好几个队列  
	30%  40%  30%
	3g      4g     3g
任务提交的时候，可以选择不同的队列来进行提交
根据提交的任务需要的资源大小不同，可以将我们的任务，划分到不同的队列下面去

第三种调度器：Fair Scheduler（公平调度器，CDH版本的hadoop默认使用的调度器）

如果没有任务提交，第一个任务过来，将集群当中的所有的资源全部给第一个任务
第二个任务来了，将第一个任务的资源划分一点出来给第二个任务，保证第二个任务也可以执行
保证每一个任务都可以公平的一起执行

一般调度器不会去更改
使用哪种调度器取决于yarn-site.xml当中的
yarn.resourcemanager.scheduler.class  这个属性的配置

关于yarn常用参数设置

第一个参数：container分配最小内存
 yarn.scheduler.minimum-allocation-mb	  1024   给应用程序container分配的最小内存

第二个参数：container分配最大内存
 yarn.scheduler.maximum-allocation-mb	  8192	给应用程序container分配的最大内存

第三个参数：每个container的最小虚拟内核个数
yarn.scheduler.minimum-allocation-vcores	  1	  每个container默认给分配的最小的虚拟内核个数

第四个参数：每个container的最大虚拟内核个数
yarn.scheduler.maximum-allocation-vcores	  32  每个container可以分配的最大的虚拟内核的个数

第五个参数：nodeManager可以分配的内存大小
yarn.nodemanager.resource.memory-mb   8192  nodemanager可以分配的最大内存大小，默认8192Mb
在我们浏览yarn的管理界面的时候会发现一个问题

我们可以在yarn-site.xml当中修改以下两个参数来改变默认值

定义每台机器的内存使用大小
yarn.nodemanager.resource.memory-mb	8192
定义每台机器的虚拟内核使用大小
yarn.nodemanager.resource.cpu-vcores	8

定义交换区空间可以使用的大小（交换区空间就是讲一块硬盘拿出来做内存使用）
这里指定的是nodemanager的n内存的2.1倍
一个真实cpu可以虚拟化出8个虚拟cpu
yarn.nodemanager.vmem-pmem-ratio	2.1

分享到：

赞(0) 打赏