一、自定义InputFormat
InputFormat是输入流，在前面的例子中使用的是文件输入输出流FileInputFormat和FileOutputFormat，而FileInputFormat和FileOutputFormat它们默认使用的是继承它们的子类TextInputFormat和TextOutputFormat，以Text的方式去读取数据。
当我们遇到许多小文件，要将他们整理合成为一个文件SequenceFile(存储了多个小文件)，且文件内的存储格式为：文件路径+文件内容，这时我们可以通过封装自定义的InputFormat输入流来实现需求。
思路如下：
1.自定义FuncFileInputFormat类继承FileInputFormat(参数类型为NullWritable和BytesWritable)，并重写isSplitable和createRecordReader方法；
2.isSplitable方法中return false即可表示不切割，createRecordReader方法中要返回一个RecordReader类，这是我们要自定义的对输入文件的业务逻辑，所以创建FuncRecordReader类；
3.FuncRecordReader类继承RecordReader类，参数类型同为NullWritable和BytesWritable，重写initialize、nextKeyValue、getCurrentKey、getCurrentValue、getProcess、close方法；
4.Mapper：初始化setup方法，通过context拿到切片、获取路径、将路径写入定义的全局变量Text t，然后在map阶段将t和value输出到reducer；
5.Reducer：遍历values，输出key，value；
6.Driver：在设置完Mapper和Reducer类后，添加设置setInputFormatClass为FuncFileInputFormat、设置setOutputFormatClass为SequenceFileOutputFormat。
代码如下：
/*** @author: PrincessHug* @date: 2019/3/29, 20:49* @Blog: https://www.cnblogs.com/HelloBigTable/*/
public class FuncFileInputFormat extends FileInputFormat<NullWritable, BytesWritable> {@Overrideprotected boolean isSplitable(JobContext context, Path filename) {return false;}@Overridepublic RecordReader<NullWritable, BytesWritable> createRecordReader(InputSplit inputSplit, TaskAttemptContext taskAttemptContext) throws IOException, InterruptedException {FuncRecordReader recordReader = new FuncRecordReader();return recordReader;}
}public class FuncRecordReader extends RecordReader<NullWritable, BytesWritable> {boolean isProcess = false;FileSplit split;Configuration conf;BytesWritable value = new BytesWritable();//初始化@Overridepublic void initialize(InputSplit inputSplit, TaskAttemptContext taskAttemptContext) throws IOException, InterruptedException {//初始化切片文件this.split = (FileSplit) inputSplit;//初始化配置信息conf = taskAttemptContext.getConfiguration();}//获取下一个文件@Overridepublic boolean nextKeyValue() throws IOException, InterruptedException {if (!isProcess){//根据切片的长度来创建缓冲区byte[] buf = new byte[(int) split.getLength()];FSDataInputStream fis = null;FileSystem fs = null;try {//获取路径Path path = split.getPath();//根据路径获取文件系统fs = path.getFileSystem(conf);//拿到输入流fis = fs.open(path);//数据拷贝IOUtils.readFully(fis,buf,0,buf.length);//拷贝缓存到最终的输出value.set(buf,0,buf.length);} catch (IOException e) {e.printStackTrace();} finally {IOUtils.closeStream(fis);IOUtils.closeStream(fs);}isProcess = true;return true;}return false;}@Overridepublic NullWritable getCurrentKey() throws IOException, InterruptedException {return NullWritable.get();}@Overridepublic BytesWritable getCurrentValue() throws IOException, InterruptedException {return value;}@Overridepublic float getProgress() throws IOException, InterruptedException {return 0;}@Overridepublic void close() throws IOException {}
}public class SequencceFileMapper extends Mapper<NullWritable, BytesWritable, Text,BytesWritable> {Text t = new Text();@Overrideprotected void setup(Context context) throws IOException, InterruptedException {//拿到切片信息FileSplit split = (FileSplit) context.getInputSplit();//路径Path path = split.getPath();//即带路径有待名称t.set(path.toString());}@Overrideprotected void map(NullWritable key, BytesWritable value, Context context) throws IOException, InterruptedException {context.write(t,value);}
}public class SequenceFileReducer extends Reducer<Text, BytesWritable,Text,BytesWritable> {@Overrideprotected void reduce(Text key, Iterable<BytesWritable> values, Context context) throws IOException, InterruptedException {for (BytesWritable v:values){context.write(key,v);}}
}public class SequenceFileDriver {public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException {//1.获取job信息Configuration conf = new Configuration();Job job = Job.getInstance(conf);//2.获取Jar包job.setJarByClass(SequenceFileDriver.class);//3.获取Mapper、Redcuer类job.setMapperClass(SequencceFileMapper.class);job.setReducerClass(SequenceFileReducer.class);//4.设置自定义读取方法job.setInputFormatClass(FuncFileInputFormat.class);//5.设置默认的输出方式job.setOutputFormatClass(SequenceFileOutputFormat.class);//6.获取Mapper输出数据类型job.setMapOutputKeyClass(Text.class);job.setMapOutputValueClass(BytesWritable.class);//7.获取Reducer输出数据类型job.setOutputKeyClass(Text.class);job.setOutputValueClass(BytesWritable.class);//8.设置输入存在的路径与处理后的结果路径FileInputFormat.setInputPaths(job,new Path("G:\\mapreduce\\inputformat\\in"));FileOutputFormat.setOutputPath(job,new Path("G:\\mapreduce\\inputformat\\out"));//9.提交任务if (job.waitForCompletion(true)){System.out.println("运行完成！");}else {System.out.println("运行失败！");}}
}
二、自定义OutputFormat
需求：目前我们有一个网站ip的文件，每行都有一个网站的ip地址，要求我们将含有“www.baidu.com”的ip地址取出放入一个结果文件，其他的地址放入另一个结果文件。
思路：1.首先Mapper、Reduer就是简单的读取数据、写出数据；
2.自定义FuncFileOutputFormat，重写它的getRecordWriter方法，返回一个FIleRecordWriter对象，这里我们再定义一个FileRecordWriter，重写FileRecordWriter、write、close方法；
3.Driver:再设置Reducer输出后添加设置setOutputFormatClass为我们自定义的FuncFileOutputFormat即可；
代码如下：
/*** @author: PrincessHug* @date: 2019/3/30, 14:44* @Blog: https://www.cnblogs.com/HelloBigTable/*/
public class FileMapper extends Mapper<LongWritable, Text, IntWritable, Text> {@Overrideprotected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {context.write(new IntWritable(1),new value);}
}public class FileReducer extends Reducer<IntWritable, Text,Text,NullWritable> {@Overrideprotected void reduce(IntWritable key, Iterable<Text> values, Context context) throws IOException, InterruptedException {for (Text k:values){String s = k.toString() + "\n";context.write(new Text(s),NullWritable.get());}}
}public class FuncFileOutputFormat extends FileOutputFormat<Text, NullWritable> {@Overridepublic RecordWriter<Text, NullWritable> getRecordWriter(TaskAttemptContext taskAttemptContext) throws IOException, InterruptedException {return new FileRecordWriter(taskAttemptContext);}
}public class FileRecordWriter extends RecordWriter<Text, NullWritable> {Configuration conf = null;FSDataOutputStream baidulog = null;FSDataOutputStream otherlog = null;//定义数据输出路径public FileRecordWriter(TaskAttemptContext taskAttemptContext) throws IOException {//获取配置信息和文件系统conf = taskAttemptContext.getConfiguration();FileSystem fs = FileSystem.get(conf);//定义输出路径itstarlog = fs.create(new Path("G:\\mapreduce\\outputformat\\out\\itstart\\baidu.logs"));otherlog = fs.create(new Path("G:\\mapreduce\\outputformat\\out\\other\\other.logs"));}//数据输出@Overridepublic void write(Text key, NullWritable value) throws IOException, InterruptedException {if (key.toString().contains("baidu")){baidulog.write(key.getBytes());}else {otherlog.write(key.getBytes());}}//关闭资源@Overridepublic void close(TaskAttemptContext taskAttemptContext) throws IOException, InterruptedException {if (itstarlog != null){itstarlog.close();}if (otherlog != null){otherlog.close();}}
}public class FileDriver {public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException {//配置、jobConfiguration conf = new Configuration();Job job = Job.getInstance(conf);//jar包job.setJarByClass(FileDriver.class);//Mapper、Reducerjob.setMapperClass(FileMapper.class);job.setReducerClass(FileReducer.class);//Mapper输出job.setMapOutputKeyClass(IntWritable.class);job.setMapOutputValueClass(Text.class);//Reudcer输出job.setOutputKeyClass(Text.class);job.setOutputValueClass(NullWritable.class);//自定义输出类job.setOutputFormatClass(FuncFileOutputFormat.class);//文件输入输出流FileInputFormat.setInputPaths(job,new Path("G:\\mapreduce\\outputformat\\in"));FileOutputFormat.setOutputPath(job,new Path("G:\\mapreduce\\outputformat\\out"));//提交任务if (job.waitForCompletion(true)){System.out.println("运行完成！");}else {System.out.println("运行失败！");}}
}
