MapReduce之自定义OutputFormat
时间:2022-07-23
本文章向大家介绍MapReduce之自定义OutputFormat,主要内容包括其使用实例、应用技巧、基本知识点总结和需要注意事项,具有一定的参考价值,需要的朋友可以参考一下。
OutputFormat接口实现类
OutputFormat
是MapReduce
输出的基类,所有实现MapReduce
输出都实现了OutputFormat
接口。下面介绍几种常见的OutputFormat实现类。
- 文本输出
TextoutputFormat
默认的输出格式是TextOutputFormat,它把每条记录写为文本行。它的键和值可以是任意类型,因为TextOutputFormat调用toString()方法把它们转换为字符串。 -
SequenceFileOutputFormat
将SecquenceFileOutputFormat输出作为后续MapReduce任务的输入,这便是一种好的输出格式,因为它的格式紧凑,很容易被压缩。 - 自定义OutputFormat 根据用户需求,自定义实现输出。
自定义OutputFormat使用场景及步骤
使用场景
- 为了实现控制最终文件的输出路径和输出格式,可以自定义OutputFormat。 例如:要在一个MapReduce程序中根据数据的不同输出两类结果到不同目录,这类灵活的输出需求可以通过自定义OutputFormat来实现。
- 自定义OutputFormat步骤
(1)自定义一个类继承
FileOutputFormat
。 (2)改写RecordWriter
,具体改写输出数据的方法write()
。
自定义OutputFormat 案例实操
需求 过滤输入的log日志,包含atguigu的网站输出到e:/atguigu.log,不包含atguigu的网站输出到e:/other.log。
输入数据
什么时候需要Reduce ①合并 ②需要对数据排序
所以本案例不需要Reduce阶段,key-value不需要实现序列化
CustomOFMapper.java
public class CustomOFMapper extends Mapper<LongWritable, Text, String, NullWritable>{
@Override
protected void map(LongWritable key, Text value, Mapper<LongWritable, Text, String, NullWritable>.Context context) throws IOException, InterruptedException {
String content = value.toString();
//value不需要,但是不能用Null这个关键字,要使用NullWritable对象
context.write(content+"rn", NullWritable.get());
}
}
MyOutPutFormat.java
public class MyOutPutFormat extends FileOutputFormat<String, NullWritable>{
@Override
public RecordWriter<String, NullWritable> getRecordWriter(TaskAttemptContext job)
throws IOException, InterruptedException {
return new MyRecordWriter(job);//传递job对象,才能在RecordWriter中获取配置
}
}
MyRecordWriter.java
public class MyRecordWriter extends RecordWriter<String, NullWritable> {
private Path atguiguPath=new Path("e:/atguigu.log");
private Path otherPath=new Path("e:/other.log");
private FSDataOutputStream atguguOS ;
private FSDataOutputStream otherOS ;
private FileSystem fs;
private TaskAttemptContext context;
public MyRecordWriter(TaskAttemptContext job) throws IOException {
context=job;
Configuration conf = job.getConfiguration();
fs=FileSystem.get(conf);
atguiguOS = fs.create(atguiguPath);
otherOS = fs.create(otherPath);
}
// 将key-value写出到文件
@Override
public void write(String key, NullWritable value) throws IOException, InterruptedException {
if (key.contains("atguigu")) {
atguguOS.write(key.getBytes());//写到atguigu.log
//统计输出的含有atguigu字符串的key-value个数
context.getCounter("MyCounter", "atguiguCounter").increment(1);
}else {
otherOS.write(key.getBytes());//写到other.log
context.getCounter("MyCounter", "otherCounter").increment(1);
}
}
// 关闭流
@Override
public void close(TaskAttemptContext context) throws IOException, InterruptedException {
if (atguguOS != null) {
IOUtils.closeStream(atguguOS);
}
if (otherOS != null) {
IOUtils.closeStream(otherOS);
}
if (fs != null) {
fs.close();
}
}
}
CustomOFDriver.java
public class CustomOFDriver {
public static void main(String[] args) throws Exception {
Path inputPath=new Path("e:/mrinput/outputformat");
Path outputPath=new Path("e:/mroutput/outputformat");
//作为整个Job的配置
Configuration conf = new Configuration();
//保证输出目录不存在
FileSystem fs=FileSystem.get(conf);
if (fs.exists(outputPath)) {
fs.delete(outputPath, true);
}
// ①创建Job
Job job = Job.getInstance(conf);
//重点,设置为自定义的输出格式
job.setJarByClass(CustomOFDriver.class);
// ②设置Job
// 设置Job运行的Mapper,Reducer类型,Mapper,Reducer输出的key-value类型
job.setMapperClass(CustomOFMapper.class);
// 设置输入目录和输出目录
FileInputFormat.setInputPaths(job, inputPath);
FileOutputFormat.setOutputPath(job, outputPath);
// 设置输入和输出格式
job.setOutputFormatClass(MyOutPutFormat.class);
// 取消reduce阶段。设置为0,默认为1
job.setNumReduceTasks(0);
// ③运行Job
job.waitForCompletion(true);
}
}
输出文件:
- ASM 翻译系列第二十一弹:ASM Attributes Directory
- Linux:sed命令详解
- ASM 翻译系列第二十二弹:ASM Internal ASM file number 8
- Ghost安装Win7/XP后自动恢复IP的批处理
- ASM 翻译系列第二十三弹:ASM Internal ASM files number 12 and 254
- Shell脚本的简单排错法及调试程序bashdb
- ASM 翻译系列第二十四弹:ASM Internal ASM files number 10 and 11
- Redhat设置NFS挂载的简单步骤
- WordPress后台首页显示RSS错误的解决办法
- ASM 翻译系列第十九弹:ASM Internal ASM Continuing Operations Directory
- ASM 翻译系列第二十五弹:ASM 高级知识 When will my rebalance complete
- ASM 翻译系列第二十六弹:ASM 高级知识 Where is my data
- LVS+Keepalived负载均衡主备&双主架构全攻略
- ASM 翻译系列第二十八弹:ASM INTERNAL Partnership and Status Table
- JavaScript 教程
- JavaScript 编辑工具
- JavaScript 与HTML
- JavaScript 与Java
- JavaScript 数据结构
- JavaScript 基本数据类型
- JavaScript 特殊数据类型
- JavaScript 运算符
- JavaScript typeof 运算符
- JavaScript 表达式
- JavaScript 类型转换
- JavaScript 基本语法
- JavaScript 注释
- Javascript 基本处理流程
- Javascript 选择结构
- Javascript if 语句
- Javascript if 语句的嵌套
- Javascript switch 语句
- Javascript 循环结构
- Javascript 循环结构实例
- Javascript 跳转语句
- Javascript 控制语句总结
- Javascript 函数介绍
- Javascript 函数的定义
- Javascript 函数调用
- Javascript 几种特殊的函数
- JavaScript 内置函数简介
- Javascript eval() 函数
- Javascript isFinite() 函数
- Javascript isNaN() 函数
- parseInt() 与 parseFloat()
- escape() 与 unescape()
- Javascript 字符串介绍
- Javascript length属性
- javascript 字符串函数
- Javascript 日期对象简介
- Javascript 日期对象用途
- Date 对象属性和方法
- Javascript 数组是什么
- Javascript 创建数组
- Javascript 数组赋值与取值
- Javascript 数组属性和方法
- Docker快速搭建WordPress博客网站
- Elasticsearch入门,这一篇就够了
- 玩转 lua in Redis
- 用docker快速搭建wordpress博客
- 生产环境优雅的重启基于Nginx、Tornado的Web服务进程
- nginx记录post body/payload数据 日志用打印出结果
- nginx的location、rewrite玩法详解
- Redis Lua脚本调试
- 安装elasticsearch 5.x, 6.x 常见问题(坑)的解决
- 轻松搞懂elasticsearch概念
- 深入Golang调度器之GMP模型
- 深入Golang之sync.Pool详解
- 一次对电视盒子的漏洞分析
- MQ选型之RabbitMQ
- httprouter与 fasthttp 的性能对比