MapReduce的API介绍

Posted 2022-03-18 月疯

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写入流程开始的时候