【Hadoop】9.MapReduce框架原理-OutputFormat数据输出

本文主要是介绍【Hadoop】9.MapReduce框架原理-OutputFormat数据输出，希望对大家解决编程问题提供一定的参考价值，需要的开发者们随着小编来一起学习吧！

在前面，我们知道了多种输入模式，输出也一样。OutputFormat是MapReduce输出的基类，所有实现MapReduce输出都实现了OutputFormat接口。

OutputFormat 接口实现类

文本输出TextOutputFormat
默认的输出格式是TextOutputFormat ，它把每条记录写为文本行。它的键和值可以是任意类型，因为TextOutputFormat调用toString（）方法把它们转换为字符串
SequenceFileOutputFormat
将SequenceFileOutputFormat输出作为后续MapReduce任务的输入，这便是一种好的输出格式，因为他的格式紧凑，很容易被压缩。
自定义OutputFormat
根据自己的需求，自定义实现。

自定义OutputFormat实现过程

步骤：

自定义一个类继承FileOutputFormat
改写RecordWriter，具体改写输出数据的方法write()

示例：
在这里插入图片描述

CustomOutputDriver

package com.xing.MapReduce.CustomOutputFormat;import com.xing.MapReduce.Flowsum.FlowBean;
import com.xing.MapReduce.Flowsum.FlowBeanMapper;
import com.xing.MapReduce.Flowsum.FlowDriver;
import com.xing.MapReduce.Flowsum.FlowReducer;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.NullWritable;
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 java.io.IOException;public class CustomOutputDriver {public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException {System.setProperty("hadoop.home.dir", "E:\\hadoop-2.7.1");String in = "E:\\hdfs\\data\\customout\\input\\demo.txt";String out = "E:\\hdfs\\data\\customout\\out";Path inPath = new Path(in);Path outPath = new Path(out);Configuration configuration = new Configuration();FileSystem fs = FileSystem.get(configuration);if (fs.exists(outPath)) {if (fs.delete(outPath, true)){System.out.println("success delete outfile");}}Job job = Job.getInstance(configuration);job.setJobName("CustomOutput");job.setJarByClass(CustomOutputDriver.class);job.setMapperClass(CustomOutputMapper.class);job.setReducerClass(CustomOutputReduce.class);job.setMapOutputKeyClass(Text.class);job.setMapOutputValueClass(NullWritable.class);job.setOutputKeyClass(Text.class);job.setOutputValueClass(NullWritable.class);// 设置自定义FileOutputFormat类job.setOutputFormatClass(CustomFileOutputFormat.class);FileInputFormat.setInputPaths(job, inPath);FileOutputFormat.setOutputPath(job, outPath);boolean rel = job.waitForCompletion(true);if (rel) {System.out.println("success");}}}

CustomOutputMapper

package com.xing.MapReduce.CustomOutputFormat;import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;import java.io.IOException;public class CustomOutputMapper extends Mapper<LongWritable,Text,Text,NullWritable> {@Overrideprotected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {context.write(value,NullWritable.get() );}
}

CustomOutputReduce

package com.xing.MapReduce.CustomOutputFormat;import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;import java.io.IOException;public class CustomOutputReduce extends Reducer<Text,NullWritable,Text,NullWritable> {@Overrideprotected void reduce(Text key, Iterable<NullWritable> values, Context context) throws IOException, InterruptedException {for (NullWritable value : values) {context.write(key,value );}}
}

CustomFileOutputFormat

package com.xing.MapReduce.CustomOutputFormat;import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.RecordWriter;
import org.apache.hadoop.mapreduce.TaskAttemptContext;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;import java.io.IOException;public class CustomFileOutputFormat extends FileOutputFormat<Text,NullWritable> {/***  返回自定义的Writer* @param context* @return* @throws IOException* @throws InterruptedException*/public RecordWriter<Text, NullWritable> getRecordWriter(TaskAttemptContext context) throws IOException, InterruptedException {return new CustomRecordWriter(context);}
}

CustomRecordWriter

package com.xing.MapReduce.CustomOutputFormat;import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FSDataOutputStream;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IOUtils;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.RecordWriter;
import org.apache.hadoop.mapreduce.TaskAttemptContext;import java.io.IOException;public class CustomRecordWriter extends RecordWriter<Text, NullWritable> {private FileSystem fs;private Configuration conf;private FSDataOutputStream fos1;private FSDataOutputStream fos2;CustomRecordWriter(){}CustomRecordWriter(TaskAttemptContext context) {// 初始化一些属性try {conf = context.getConfiguration();fs = FileSystem.get(conf);fos1 = fs.create(new Path("E:\\hdfs\\data\\customout\\output\\http.txt"));fos2 = fs.create(new Path("E:\\hdfs\\data\\customout\\output\\other.txt"));} catch (IOException e) {e.printStackTrace();}}/***  判断key值是否含有HTTP，有则输出到http.txt 其他都输出到other.txt文件* @param text* @param nullWritable* @throws IOException* @throws InterruptedException*/public void write(Text text, NullWritable nullWritable) throws IOException, InterruptedException {if (text.toString().toUpperCase().contains("HTTP")){System.out.println("yes");// \r\n 为windows的换行 fos1.write(text.toString().concat("\r\n").getBytes());}else {System.out.println("no");fos2.write(text.toString().concat("\r\n").getBytes());}}/***  关闭流* @param taskAttemptContext* @throws IOException* @throws InterruptedException*/public void close(TaskAttemptContext taskAttemptContext) throws IOException, InterruptedException {IOUtils.closeStream(fos1);IOUtils.closeStream(fos2);}
}