====
案例：自定义outputFormat--将订单的好评与差评进行区分开来
2.1 需求
现在有一些订单的评论数据，需求，将订单的好评与差评进行区分开来，将最终的数据分开到不同的文件夹下面去，数据内容参见资料文件夹，其中数据第九个字段表示好评，中评，差评。0：好评，1：中评，2：差评
2.2 分析
程序的关键点是要在一个mapreduce程序中根据数据的不同输出两类结果到不同目录，这类灵活的输出需求可以通过自定义outputformat来实现
2.3 实现
实现要点：
1、 在mapreduce中访问外部资源
2、 自定义outputformat，改写其中的recordwriter，改写具体输出数据的方法write()
k1: LongWritable; v1: Text
k2: Text, 一行文本内容; v2: NullWritable
k3: Text; v3: NullWritable
代码实现：
package cn.itcast.demo4.outputformat;
import org.apache.hadoop.fs.FSDataOutputStream;
import org.apache.hadoop.fs.FileSystem;
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 MyRecordWriter extends RecordWriter<Text,NullWritable> {
private FSDataOutputStream goodStream;
private FSDataOutputStream badStream;
public MyRecordWriter(FSDataOutputStream goodStream,FSDataOutputStream badStream){
this.goodStream = goodStream;
this.badStream = badStream;
}
@Override
public void write(Text text, NullWritable nullWritable) throws IOException, InterruptedException {
String[] split = text.toString().split("\t");
//获取我们的评论状态
String s = split[9];
if(Integer.parseInt(s) <=1){
//好评与中评，写入到一个文件夹里面去
goodStream.write(text.toString().getBytes());
goodStream.write("\r\n".getBytes());
goodStream.flush();
}else{
//差评，写入到一个文件夹里面去
badStream.write(text.toString().getBytes());
badStream.write("\r\n".getBytes());
badStream.flush();
}
}
@Override
public void close(TaskAttemptContext taskAttemptContext) throws IOException, InterruptedException {
this.goodStream.close();
this.badStream.close();
}
}
第一步：自定义一个outputformat
package cn.itcast.demo4.outputformat;
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.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 MyOutputFormat extends FileOutputFormat<Text,NullWritable> {
/**
* 重新recordWriter方法
* @param taskAttemptContext
* @return
* @throws IOException
* @throws InterruptedException
*/
@Override
public RecordWriter<Text, NullWritable> getRecordWriter(TaskAttemptContext taskAttemptContext) throws IOException, InterruptedException {
Configuration configuration = taskAttemptContext.getConfiguration();
//获取到了文件系统客户端
FileSystem fileSystem = FileSystem.get(configuration);
Path goodCommentPath = new Path("file:///F:\\传智播客大数据离线阶段课程资料\\5、大数据离线第五天\\自定义outputformat\\googcomment\\good.txt");
Path badCommentPath = new Path("file:///F:\\传智播客大数据离线阶段课程资料\\5、大数据离线第五天\\自定义outputformat\\badcomment\\bad.txt");
//这个输出流，就是用于我们将好评数据写出去
FSDataOutputStream goodStream = fileSystem.create(goodCommentPath);
FSDataOutputStream badStream = fileSystem.create(badCommentPath);
MyRecordWriter myRecordWriter = new MyRecordWriter(goodStream,badStream);
return myRecordWriter;
}
}
package cn.itcast.demo4.outputformat;
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 MyOutputFormatMapper extends Mapper<LongWritable,Text,Text,NullWritable> {
@Override
protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
context.write(value,NullWritable.get());
}
}
package cn.itcast.demo4.outputformat;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;
import java.io.IOException;
public class MyOutputFormatReducer extends Reducer<Text,NullWritable,Text,NullWritable> {
@Override
protected void reduce(Text key, Iterable<NullWritable> values, Context context) throws IOException, InterruptedException {
context.write(key,NullWritable.get());
}
}
package cn.itcast.demo4.outputformat;
import cn.itcast.demo3.mergefile.MyInputFormat;
import cn.itcast.demo3.mergefile.MyInputMappr;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.BytesWritable;
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.TextInputFormat;
import org.apache.hadoop.mapreduce.lib.output.SequenceFileOutputFormat;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;
public class MyOutputFormatMain extends Configured implements Tool {
@Override
public int run(String[] args) throws Exception {
Job job = Job.getInstance(super.getConf(), "myoutputformat");
job.setInputFormatClass(TextInputFormat.class);
TextInputFormat.addInputPath(job,new Path("file:///F:\\传智播客大数据离线阶段课程资料\\5、大数据离线第五天\\自定义outputformat\\input"));
job.setMapperClass(MyOutputFormatMapper.class);
job.setMapOutputKeyClass(Text.class);
job.setMapOutputValueClass(NullWritable.class);
/**
* 我们这里没有reduce的程序，但是我们输出的数据类型是SequenceFile类型的
* SequenceFile输出的数据类型，默认找的是以下这两个设置
*/
job.setReducerClass(MyOutputFormatReducer.class);
job.setOutputKeyClass(Text.class);
job.setOutputValueClass(NullWritable.class);
job.setOutputFormatClass(MyOutputFormat.class);
MyOutputFormat.setOutputPath(job,new Path("file:///F:\\传智播客大数据离线阶段课程资料\\5、大数据离线第五天\\自定义outputformat\\myoutput_check"));
boolean b = job.waitForCompletion(true);
return b?0:1;
}
public static void main(String[] args) throws Exception {
int run = ToolRunner.run(new Configuration(), new MyOutputFormatMain(), args);
System.exit(run);
}
}
====