MapReduce的API介绍

Posted 月疯

tags:

篇首语:本文由小常识网(cha138.com)小编为大家整理,主要介绍了MapReduce的API介绍相关的知识,希望对你有一定的参考价值。

MapReduce自带的InputFormat介绍:

 

SwquenceFile文件,<K,V>形式的二进制文件:
package squencefile;


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.IOUtils;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.SequenceFile;
import org.apache.hadoop.io.SequenceFile.Writer;
import org.apache.hadoop.io.Text;

import java.io.IOException;

public class GenerateSwquenceFile 
    public static void main(String[] args) throws IOException 
        //1、sequenceFile文件是通过SequenceFile类生成的
        //createWrite方法参数:conf:hadoop配置项,name:文件名
        //keyClass:key的数据类型;valClass:值得数据类型
        //指定文件名称
        Writer.Option name =Writer.file(new Path("file:/e:sf"));
        //指定key的类型
        Writer.Option keyClass=Writer.keyClass(LongWritable.class);
        //指定value的类型
        Writer.Option valClass = Writer.valueClass(Text.class);
        //hadoop配置项
        Configuration conf=new Configuration();
        //创建输出流
        Writer writer= SequenceFile.createWriter(conf,name,keyClass,valClass);
        //
        FileSystem fs =FileSystem.get(conf);
        FSDataInputStream in=fs.open(new Path("file:/e:/words.txt"));
        String line=null;
        Long num=0L;
        while((line = in.readLine() )!=null)
            ++num;
            writer.append(new LongWritable(num),new Text(line));
        
        //关闭输出流
        IOUtils.closeStream(writer);
    

SequenceFileInputFormat的使用:
//输入文件
        FileInputFormat.addInputPath(job,new Path(args[0]));
        //指定以SequenceFileInputFormat处理sequenceFIle文件
        job.setInputFormatClass(SequenceFileInputFormat.class);
        //编写mapper处理逻辑
        job.setMapperClass(MyMapper.class);
        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(LongWritable.class);

inputSplite原理:

输入切片(InputSplite)
1、什么时候切片
    client端进行切片,且丰厚交给Yarn进行处理
2、切片中存储的内容
数据长度、数据存储位置
3、切片大小
minSize = maxminSpliteSize,mapred.min.splite.size
maxSize=mapred.max.splite.size
spliteSize=maxminSize,minmaxSize,blockSize
注意:minSpliteSize大小默认为1B;mapper数:文件大小/spliteSize
4、切片数量(mapper进程数量)
总文件大小/切片的大小 

reduce个数设置:

 

outPutFormat输出格式:

 

注意输入和输出是针对内存和磁盘的,out:磁盘,in:内存。

 指定文件输出格式:

//序列化问价的压缩类型:None、Block、Record
        SequenceFileOutputFormat.setOutputCompressionType(job, SequenceFile.CompressionType.BLOCK);
        //default、gzip、lz4
        SequenceFileOutputFormat.setOutputCompressorClass(job, DefaultCodec.class);
        //格式化序列化输出
        job.setOutputFormatClass(SequenceFileOutputFormat.class);

partitioner分区器:
在mapper和reduce处理逻辑之间,shuffle写入流程开始的时候

 

Shuffle流程介绍: 

shuffle流程架构:

 

 

 

 

 

 

 

shuffler读取流程: 

 

 

 

以上是关于MapReduce的API介绍的主要内容,如果未能解决你的问题,请参考以下文章

为啥我们需要知道在 mapreduce 中正在处理的文件的名称?

通过 MapReduce 读取与特定模式匹配的目录中的文件并输出各个文件的名称

Java Mapreduce - 获取匹配文件的名称并打印到输出文件

使用 MapReduce API 通过 Gzip 压缩复制 HDFS 内的文件

hadoop map中获取文件/切片名称

大数据开发 | MapReduce介绍