【Hadoop】5.MapReduce框架原理-自定义InputFormat

2023-11-09 13:32

本文主要是介绍【Hadoop】5.MapReduce框架原理-自定义InputFormat,希望对大家解决编程问题提供一定的参考价值,需要的开发者们随着小编来一起学习吧!

说明

自定义InputFormat一般应用于hadoop自带的InputFormat类型不能满足某个应用场景中,需要我们自定义来解决

自定义步骤

自定义Inputformat步骤如下:

  1. 自定义一个类继承InputFormat
  2. 改写RecordReader,实现一次读取一个完成的文件封装为KV
  3. 在输出时使用SequenceFileOutPutFormat输出合并文件SequenceFile文件

(SequenceFile文件是hadoop用来存储二进制形式的key-value对的文件格式,SequenceFile里面存储着很多小文件,存储的形式为文件路径+名称为key,文件内容为value)。

示例操作

步骤:

  1. 自定义一个类继承FileInputFormat
    a. 重写isSpliable()方法,返回false 设置为文件不可分割
    b. 重写createRecordReader(),创建自定义的RecordReader对象,并初始化
  2. 改写RecordReader,实现一次读取一个完整的文件封装成为KV
    a. 采用IO流一次读取一个文件输出到value中,因为设置了不可分割所以整个文件都封装到了value中
    b. 获取文件路径信息+名称作为key
  3. 设置Driver
    a.

代码:
在这里插入图片描述

CustomerInputFormate自定义InputFormat
package com.xing.MapReduce.InputFormatSequenceFile;import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.BytesWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.InputSplit;
import org.apache.hadoop.mapreduce.JobContext;
import org.apache.hadoop.mapreduce.RecordReader;
import org.apache.hadoop.mapreduce.TaskAttemptContext;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;import java.io.IOException;/***  自定义InputFormat*/
public class CustomerInputFormate extends FileInputFormat<Text,BytesWritable> {/***  新建自定义的RecordReader* @param inputSplit* @param taskAttemptContext* @return* @throws IOException* @throws InterruptedException*/public RecordReader<Text, BytesWritable> createRecordReader(InputSplit inputSplit, TaskAttemptContext taskAttemptContext) throws IOException, InterruptedException {CustomerRecordReader reader = new CustomerRecordReader();reader.initialize(inputSplit,taskAttemptContext );return reader;}/***  设置为不可分割* @param context* @param filename* @return*/@Overrideprotected boolean isSplitable(JobContext context, Path filename) {return false;}
}
CustomerRecordReader 自定义RecordReader
package com.xing.MapReduce.InputFormatSequenceFile;import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FSDataInputStream;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.BytesWritable;
import org.apache.hadoop.io.IOUtils;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.InputSplit;
import org.apache.hadoop.mapreduce.RecordReader;
import org.apache.hadoop.mapreduce.TaskAttemptContext;
import org.apache.hadoop.mapreduce.lib.input.FileSplit;import java.io.IOException;/***  自定义RecordReader*/
public class CustomerRecordReader extends RecordReader<Text, BytesWritable> {// 分片信息FileSplit split;Configuration configuration;Text k = new Text();BytesWritable v = new BytesWritable();boolean isProgress = true;//初始化方法public void initialize(InputSplit inputSplit, TaskAttemptContext taskAttemptContext) throws IOException, InterruptedException {this.split = (FileSplit) inputSplit;this.configuration = taskAttemptContext.getConfiguration();}//核心业务处理(对key和value的封装)public boolean nextKeyValue() throws IOException, InterruptedException {if (isProgress){//1. 获取fsPath path = split.getPath();System.out.println("@@@@@@@@@@@@@@@@"+split.getLength());System.out.println("$$$$$$$$$$$$$$$$"+split);FileSystem fileSystem = path.getFileSystem(configuration);//2. 获取输入流FSDataInputStream fis = fileSystem.open(path);//3. 拷贝byte[] bytes = new byte[(int) split.getLength()];IOUtils.readFully(fis, bytes,0 ,bytes.length);//4. 填充k-vv.set(bytes,0 ,bytes.length);k.set(path.toString());//5. 收尾IOUtils.closeStream(fis);isProgress = false;return true;}return false;}// 获取kay值public Text getCurrentKey() throws IOException, InterruptedException {return k;}// 获取value值public BytesWritable getCurrentValue() throws IOException, InterruptedException {return v;}public float getProgress() throws IOException, InterruptedException {return 0;}public void close() throws IOException {}
}
SequenceFileMapper
package com.xing.MapReduce.InputFormatSequenceFile;import org.apache.hadoop.io.BytesWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;import java.io.IOException;public class SequenceFileMapper extends Mapper<Text,BytesWritable,Text,BytesWritable> {@Overrideprotected void map(Text key, BytesWritable value, Context context) throws IOException, InterruptedException {// 直接输出 这里的key就是文件路径信息 value就是文本内容context.write(key,value );}}
SequenceFileReducer
package com.xing.MapReduce.InputFormatSequenceFile;import org.apache.hadoop.io.BytesWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;import java.io.IOException;public class SequenceFileReducer extends Reducer<Text,BytesWritable,Text,BytesWritable> {@Overrideprotected void reduce(Text key, Iterable<BytesWritable> values, Context context) throws IOException, InterruptedException {// 就是简单的输出内容for (BytesWritable value : values) {context.write(key,value);}}
}
SequenceFileDriver
package com.xing.MapReduce.InputFormatSequenceFile;import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.BytesWritable;
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 org.apache.hadoop.mapreduce.lib.output.SequenceFileOutputFormat;import java.io.IOException;public class SequenceFileDriver {public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException {// 获取configuration和FileSystemConfiguration configuration = new Configuration();FileSystem fs = FileSystem.get(configuration);// 设置job的名字和jar包Job job = Job.getInstance(configuration);job.setJobName("SequenceFileDriver");job.setJarByClass(SequenceFileDriver.class);// 设置job的mapper和reduce的处理类名job.setMapperClass(SequenceFileMapper.class);job.setReducerClass(SequenceFileReducer.class);// 设置输入类型和输出类型job.setInputFormatClass(CustomerInputFormate.class);job.setOutputFormatClass(SequenceFileOutputFormat.class);// 设置输出的key和输出的value类型job.setOutputKeyClass(Text.class);job.setOutputValueClass(BytesWritable.class);// 判断输出地址是否存在if (fs.exists(new Path("E:\\hdfs\\output1"))){fs.delete(new Path("E:\\hdfs\\output1"),true );}// 设置输入和输出的文件路径FileInputFormat.setInputPaths(job,new Path("E:\\hdfs\\input"));FileOutputFormat.setOutputPath(job,new Path("E:\\hdfs\\output1"));// 返回结果boolean b = job.waitForCompletion(true);System.exit(b ? 0 : -1);}
}
输入和输出

输入:
在这里插入图片描述
输出:

在这里插入图片描述

这篇关于【Hadoop】5.MapReduce框架原理-自定义InputFormat的文章就介绍到这儿,希望我们推荐的文章对编程师们有所帮助!



http://www.chinasem.cn/article/376420

相关文章

从原理到实战深入理解Java 断言assert

《从原理到实战深入理解Java断言assert》本文深入解析Java断言机制,涵盖语法、工作原理、启用方式及与异常的区别,推荐用于开发阶段的条件检查与状态验证,并强调生产环境应使用参数验证工具类替代... 目录深入理解 Java 断言(assert):从原理到实战引言:为什么需要断言?一、断言基础1.1 语

MySQL中的表连接原理分析

《MySQL中的表连接原理分析》:本文主要介绍MySQL中的表连接原理分析,具有很好的参考价值,希望对大家有所帮助,如有错误或未考虑完全的地方,望不吝赐教... 目录1、背景2、环境3、表连接原理【1】驱动表和被驱动表【2】内连接【3】外连接【4编程】嵌套循环连接【5】join buffer4、总结1、背景

Spring 框架之Springfox使用详解

《Spring框架之Springfox使用详解》Springfox是Spring框架的API文档工具,集成Swagger规范,自动生成文档并支持多语言/版本,模块化设计便于扩展,但存在版本兼容性、性... 目录核心功能工作原理模块化设计使用示例注意事项优缺点优点缺点总结适用场景建议总结Springfox 是

深度解析Spring AOP @Aspect 原理、实战与最佳实践教程

《深度解析SpringAOP@Aspect原理、实战与最佳实践教程》文章系统讲解了SpringAOP核心概念、实现方式及原理,涵盖横切关注点分离、代理机制(JDK/CGLIB)、切入点类型、性能... 目录1. @ASPect 核心概念1.1 AOP 编程范式1.2 @Aspect 关键特性2. 完整代码实

Java实现自定义table宽高的示例代码

《Java实现自定义table宽高的示例代码》在桌面应用、管理系统乃至报表工具中,表格(JTable)作为最常用的数据展示组件,不仅承载对数据的增删改查,还需要配合布局与视觉需求,而JavaSwing... 目录一、项目背景详细介绍二、项目需求详细介绍三、相关技术详细介绍四、实现思路详细介绍五、完整实现代码

一文详解Java Stream的sorted自定义排序

《一文详解JavaStream的sorted自定义排序》Javastream中的sorted方法是用于对流中的元素进行排序的方法,它可以接受一个comparator参数,用于指定排序规则,sorte... 目录一、sorted 操作的基础原理二、自定义排序的实现方式1. Comparator 接口的 Lam

Java Stream的distinct去重原理分析

《JavaStream的distinct去重原理分析》Javastream中的distinct方法用于去除流中的重复元素,它返回一个包含过滤后唯一元素的新流,该方法会根据元素的hashcode和eq... 目录一、distinct 的基础用法与核心特性二、distinct 的底层实现原理1. 顺序流中的去重

Python的端到端测试框架SeleniumBase使用解读

《Python的端到端测试框架SeleniumBase使用解读》:本文主要介绍Python的端到端测试框架SeleniumBase使用,具有很好的参考价值,希望对大家有所帮助,如有错误或未考虑完全... 目录SeleniumBase详细介绍及用法指南什么是 SeleniumBase?SeleniumBase

Spring @Scheduled注解及工作原理

《Spring@Scheduled注解及工作原理》Spring的@Scheduled注解用于标记定时任务,无需额外库,需配置@EnableScheduling,设置fixedRate、fixedDe... 目录1.@Scheduled注解定义2.配置 @Scheduled2.1 开启定时任务支持2.2 创建

Spring Boot 实现 IP 限流的原理、实践与利弊解析

《SpringBoot实现IP限流的原理、实践与利弊解析》在SpringBoot中实现IP限流是一种简单而有效的方式来保障系统的稳定性和可用性,本文给大家介绍SpringBoot实现IP限... 目录一、引言二、IP 限流原理2.1 令牌桶算法2.2 漏桶算法三、使用场景3.1 防止恶意攻击3.2 控制资源