您现在的位置是:首页 > 正文

MapReduce【自定义InputFormat】

2024-02-29 16:05:23阅读 2

MapReduce在处理小文件时效率很低,但面对大量的小文件又不可避免,这个时候就需要相应的解决方案。

默认的输入格式为TextInputFormat,对于小文件,它是按照它的父类FileInputFormat的切片机制来切片的,也就是不管一个文件多小,独占一片!对于之前的wordcount案例来说,输入目录下一共有4个文件,这将开启4个reduceTask去执行!!!

我们可以自定义InputFormat来实现小文件的合并:

需求

我们需要将三个文本文件合并为一个序列化文件

 输入

三个文本文件

输出

一个二进制序列化文件

1.自定义Inputformat类

需要实现两个方法

  1. isSplitable():是否可以切片,我们修改返回值为false不可切割。
  2. createRecordReader:返回我们自定义的RecordReader对象。
public class MyInputFormat extends FileInputFormat<Text, BytesWritable> {

    //设置文件不可切片,使得一个文件最多作为1片
    @Override
    protected boolean isSplitable(JobContext context, Path filename) {
        return false;
    }

    //设置读取文件的格式为自定义格式
    @Override
    public RecordReader<Text,BytesWritable> createRecordReader(InputSplit inputSplit, TaskAttemptContext taskAttemptContext) throws IOException, InterruptedException {
        return new MyRecordReader();
    }
}

2. 自定义RecordReader类

我们主要修改两个地方:key和value。

因为我们Mapper类中的map方法需要有四个参数,其中的KEY_IN和VALUE_IN都是由我们的RecordReader类来设置的,这里我们需要设置一下。

  1. 默认的RecordReader类的key为LongWritable类型,也就是一行数据对应的字节偏移量,这里我们设置key为我们的文件名,也就是Text类型。
  2. 默认的RecordReader类的value为Text类型,也就是一行文本,这里我们设置value为文件名key对应的文件的二进制序列,也就是BytesWritable类型。
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.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<Text,BytesWritable> {
    private Text key;
    private BytesWritable value;
    private String filename;
    private int length;
    private FileSystem fs;
    private Path path;
    private FSDataInputStream is;
    private boolean flag=true;
    @Override
    public void initialize(InputSplit inputSplit, TaskAttemptContext taskAttemptContext) throws IOException, InterruptedException {

        FileSplit fileSplit = (FileSplit) inputSplit;

        filename = fileSplit.getPath().getName();

        length = (int) fileSplit.getLength();

        path = fileSplit.getPath();

        //获取当前Job的配置对象
        Configuration conf = taskAttemptContext.getConfiguration();
        //获取当前Job使用的文件系统
        fs = FileSystem.get(conf);
        is = fs.open(path);
    }


    //文件的名称做为 key - 文件的内容分装为BytesWritable类型的 value, 返回true
    @Override
    public boolean nextKeyValue() throws IOException, InterruptedException {

        //第一次调用nextKeyValue方法
        if (flag){
            //实例化对象
            if (key==null){
                key = new Text();
            }
            if (value==null){
                value = new BytesWritable();
            }

            //赋值
            //将文件名封装到key中
            key.set(filename);
            //将文件的内容读取封装到value中
            byte[] content = new byte[ length];
            IOUtils.readFully(is,content,0,length);
            value.set(content,0,length);
            flag = false;
            return true;
        }
        //第二次调用直接执行 return false
        return false;
    }

    //返回当前读取到的key
    @Override
    public Text getCurrentKey() throws IOException, InterruptedException {
        return key;
    }

    //返回当前读取到的value
    @Override
    public BytesWritable getCurrentValue() throws IOException, InterruptedException {
        return value;
    }

    //返回读取切片的进度
    @Override
    public float getProgress() throws IOException, InterruptedException {
        return 0;
    }

    //关闭资源
    @Override
    public void close() throws IOException {
        if (is != null){
            IOUtils.closeStream(is);
        }
        if (fs != null){
            fs.close();
        }
    }
}

3. Mapper类

在我们的自定义RecordReader类中,我们已经设置输入的key为文件名,value设置为文件的二进制序列,所以这里直接将key和value写出即可,key的类型为Text,value的类型为BytesWritable。

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

4.Reducer类

public class SequenceFileReducer extends Reducer<Text, BytesWritable,Text,Text> {
    private Text OUT_VALUE = new Text();
    @Override
    protected void reduce(Text key, Iterable<BytesWritable> values, Context context) throws IOException, InterruptedException {

        String value = values.toString();
        OUT_VALUE.set(value);
        context.write(key,OUT_VALUE);

    }
}

5.启动类

import com.lyh.mapreduce.MaxTemp.MaxTempMapper;
import com.lyh.mapreduce.MaxTemp.MaxTempReducer;
import com.lyh.mapreduce.MaxTemp.MaxTempRunner;
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.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 org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;

public class MySequenceFileRunner extends Configured implements Tool {
    public static void main(String[] args) throws Exception {
        ToolRunner.run(new Configuration(),new MySequenceFileRunner(),args);
    }

    @Override
    public int run(String[] args) throws Exception {
        //1.获取job
        Configuration conf = new Configuration();
        Job job = Job.getInstance(conf, "my sequence file demo");

        //2.配置jar包路径
        job.setJarByClass(MySequenceFileRunner.class);

        //3.关联mapper和reducer
        job.setMapperClass(SequenceFileMapper.class);
        job.setReducerClass(SequenceFileReducer.class);

        //4.设置map、reduce输出的k、v类型
        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(BytesWritable.class);
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(BytesWritable.class);

        //设置切片机制为我们自定义的切片机制
        job.setInputFormatClass(MyInputFormat.class);

        //5.设置统计文件输入的路径,将命令行的第一个参数作为输入文件的路径
        FileInputFormat.setInputPaths(job,new Path("D:\\MapReduce_Data_Test\\myinputformat\\input"));
        //6.设置结果数据存放路径,将命令行的第二个参数作为数据的输出路径
        FileOutputFormat.setOutputPath(job,new Path("D:\\MapReduce_Data_Test\\myinputformat\\output1"));
        return job.waitForCompletion(true) ? 0 : 1;//verbose:是否监控并打印job的信息
    }
}

执行结果

网站文章

  • did you specify the right host or port报错处理

    根据报错信息可以看到bootstrap-kubelet.conf文件已经不存在了,导致kubelet无法启动。应该是证书过期了,才导致无法执行kubectl命令了。需要手动来更新证书文件。执行完这几步以后,在试下kubectl get node,就可以查询出来node节点信息了。集群是用kubeadm来部署的。需要执行kubeadm命令来进行更新证书。

    2024-02-29 16:05:17
  • TypeScript中的运算符和操作符

    介绍一下TypeScript中的其中常见的操作符和运算符 一、非空断言操作符 (!) 在上下文中,当类型检查器无法断定类型时,一个新的后缀表达式操作符 – !可以用于断言操作对象是非null和非und...

    2024-02-29 16:05:09
  • 查看进程与线程

    查看进程的命令

    2024-02-29 16:04:40
  • Winform控件大小随窗体缩放

    Winform控件大小随窗体缩放。

    2024-02-29 16:04:33
  • 权值矩阵

    【问题描述】   Mr_he有一个 n*m 的矩阵,并且把1~n*m 这n*m填写在这个矩阵中(注意,每个整数在矩阵出现一次,且仅出现一次)。   Mr_he同时约定一个矩阵权值等于这个子矩阵中的所有数的最小值。   现在Mr_he想知道,在给出的矩阵中,权值为i的子矩阵可能有多少种? 【输入格式】   第一行, 两个整数 N, M。接下来的 N 行, 每行 M 个整数, 表示矩阵中的元素。

    2024-02-29 16:04:25
  • php语言的语法非常灵活,网站基础知识学习

    网站技术基础知识学习asp、php、http://www.doczj.com/doc/f0a7fee0ddccda38366baf00.html、jsp介绍及优缺点比较ASP:ASP是微软(Micro...

    2024-02-29 16:04:16
  • python基础(2)

    文章目录list和truple一、list二、truple1、truple的意义2、truple的表示方法3、可变的truple程序语言的三种基本结构一、顺序二、条件练习题三、循环forwhileli...

    2024-02-29 16:03:48
  • div float style=&quot;clear:both&quot;

    aaaaaaaaaaaaaa bbbbbbbbbbbbbb 以上这两个div就是横向排列。 这个东西很重要,它...

    2024-02-29 16:03:42
  • mysql backup.sh_mysql_backup.sh

    #!/bin/bashpath=/bin:/sbin:/usr/bin:/usr/sbin:/usr/local/bin:/usr/local/sbinexportpath#数据库用户名dbuser=...

    2024-02-29 16:03:35
  • Gson设置解析date格式

    使用gson将实体类对象解析json或者用json生成实体类时,解析出来的date格式是X月 XX,XXXX, 直接在创建gson时设置时间格式: Gson gson = new GsonBuilde...

    2024-02-29 16:03:04