MapReduce默认的输出格式为TextOutputFormat，它的父类是FileOutputFormat，即按行来写，且内容写到一个文本文件中去,但是并不能满足我们实际开发中的所有需求，所以就需要我们自定义OutPutFormat。
自定义OutPutFormat
输出数据到MySQL、HBase或者Elasticsearch等存储框架中。
步骤
继承FileOutputFormat
继承RecordWriter类
重写write方法
案例
需求
过滤log日志，将包含sxau的网站输出到sxau.log，其他则输出到other.log
输入数据
www.baidu.com www.google.com www.sxau.com www.jd.com www.bing.com www.sina.com www.csdn.com www.github.com
LogMapper类
public class LogMapper extends Mapper<LongWritable, Text,Text, NullWritable> { @Override protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException { //直接写出 context.write(value,NullWritable.get()); } }
LogReducer类
public class LogReducer extends Reducer<Text, NullWritable,Text,NullWritable> { @Override protected void reduce(Text key, Iterable<NullWritable> values, Context context) throws IOException, InterruptedException { //防止相同数据丢失 for (NullWritable value : values) { context.write(key,NullWritable.get()); } } }
MyOutputFormat类
返回我们自定义的RecordWriter类
public class MyOutputFormat extends FileOutputFormat<Text, NullWritable> { @Override public RecordWriter<Text, NullWritable> getRecordWriter(TaskAttemptContext job) throws IOException, InterruptedException { //将job对象传进去,才能方便对数据流进行操作 MyRecordWriter recordWriter = new MyRecordWriter(job); return recordWriter; } }
MyRecordWriter类
核心是重写write方法
public class MyRecordWriter extends RecordWriter<Text, NullWritable> { private FSDataOutputStream sxauOut; private FSDataOutputStream otherOut; public MyRecordWriter(TaskAttemptContext job) { //创建两个流 try { FileSystem fs = FileSystem.get(job.getConfiguration()); //输出路径 sxauOut = fs.create(new Path("D:\\MapReduce_Data_Test\\myoutputformat\\sxau.log")); otherOut = fs.create(new Path("D:\\MapReduce_Data_Test\\myoutputformat\\other.log")); } catch (IOException e) { e.printStackTrace(); } } @Override public void write(Text key, NullWritable value) throws IOException, InterruptedException { String log = key.toString(); System.out.println(log); //具体的写出 if (log.contains("sxau")){ sxauOut.writeBytes(log+"\n"); }else { otherOut.writeBytes(log+"\n"); } } @Override public void close(TaskAttemptContext taskAttemptContext) throws IOException, InterruptedException { //关闭数据流 IOUtils.closeStream(sxauOut); IOUtils.closeStream(otherOut); } }
LogRunner类
设置输出格式
public class LogRunner extends Configured implements Tool { public static void main(String[] args) throws Exception { ToolRunner.run(new Configuration(),new LogRunner(),args); } @Override public int run(String[] args) throws Exception { //1.获取job Configuration conf = new Configuration(); Job job = Job.getInstance(conf, "my outputFormat"); //2.配置jar包路径 job.setJarByClass(LogRunner.class); //3.关联mapper和reducer job.setMapperClass(LogMapper.class); job.setReducerClass(LogReducer.class); //4.设置map、reduce输出的k、v类型 job.setMapOutputKeyClass(Text.class); job.setMapOutputValueClass(NullWritable.class); job.setOutputKeyClass(Text.class); job.setOutputValueClass(NullWritable.class); //设置自定义输出格式 job.setOutputFormatClass(MyOutputFormat.class); //5.设置统计文件输入的路径,将命令行的第一个参数作为输入文件的路径 FileInputFormat.setInputPaths(job,new Path("D:\\MapReduce_Data_Test\\myoutputformat\\input")); //6.设置结果数据存放路径,将命令行的第二个参数作为数据的输出路径 FileOutputFormat.setOutputPath(job,new Path("D:\\MapReduce_Data_Test\\myoutputformat\\output1")); return job.waitForCompletion(true) ? 0 : 1;//verbose：是否监控并打印job的信息 } }
运行结果
