经验首页 前端设计 程序设计 Java相关 移动开发 数据库/运维 软件/图像 大数据/云计算 其他经验
当前位置：技术经验 » 大数据/云/AI » Hadoop » 查看文章
大数据-Hadoop生态(16)-MapReduce框架原理-自定义FileInputFormat
来源：cnblogs　　作者：nt杨　　时间：2018/12/12 9:42:28　　对本文有异议

 

1. 需求

将多个小文件合并成一个SequenceFile文件（SequenceFile文件是Hadoop用来存储二进制形式的key-value对的文件格式），SequenceFile里面存储着多个文件，存储的形式为文件路径+名称为key，文件内容为value

三个小文件

one.txt

  1. yongpeng weidong weinan
  2. sanfeng luozong xiaoming

two.txt

  1. shuaige changmo zhenqiang 
  2. dongli lingu xuanxuan

three.txt

  1. longlong fanfan
  2. mazong kailun yuhang yixin
  3. longlong fanfan
  4. mazong kailun yuhang yixin

 

2. 需求分析

 

3.案例代码

1) 自定义RecordReader

  1. package com.nty.inputformat;
  3. import org.apache.hadoop.fs.FSDataInputStream;
  4. import org.apache.hadoop.fs.FileSystem;
  5. import org.apache.hadoop.fs.Path;
  6. import org.apache.hadoop.io.BytesWritable;
  7. import org.apache.hadoop.io.IOUtils;
  8. import org.apache.hadoop.io.Text;
  9. import org.apache.hadoop.mapreduce.InputSplit;
  10. import org.apache.hadoop.mapreduce.RecordReader;
  11. import org.apache.hadoop.mapreduce.TaskAttemptContext;
  12. import org.apache.hadoop.mapreduce.lib.input.FileSplit;
  14. import java.io.IOException;
  16. /**
  17.  * author nty
  18.  * date time 2018-12-11 9:10
  19.  */
  20. public class CustomRecordReader extends RecordReader<Text, BytesWritable> {
  22.     /**
  23.      * 由于采用了FileInputFormat的输入方式,所以输入源3个文件,会分成三个切片,所以一个RecordReader只处理一个文件,一次读完
  24.      */
  25.  
  26.     //标记文件是否被读过,true表示没被读过
  27.     private boolean flag = true;
  29.     private Text key = new Text();
  30.     private BytesWritable value = new BytesWritable();
  32.     //输入流
  33.     FSDataInputStream fis;
  35.     private FileSplit fs;
  37.     /**
  38.      * 初始化方法,只调用一次
  39.      * @param split
  40.      * @param context
  41.      * @throws IOException
  42.      * @throws InterruptedException
  43.      */
  44.     public void initialize(InputSplit split, TaskAttemptContext context) throws IOException, InterruptedException {
  45.         //FileSplit是InputSplit的子类
  46.         fs = (FileSplit) split;
  48.         //获取文件路径
  49.         Path path = fs.getPath();
  51.         //获取文件系统
  52.         FileSystem fileSystem = FileSystem.get(context.getConfiguration());
  53.         //FileSystem fileSystem = path.getFileSystem(context.getConfiguration());
  55.         //开流
  56.         fis = fileSystem.open(path);
  57.     }
  59.     /**
  60.      * 读取下一组KV
  61.      * @return 读到了返回true,反之返回false
  62.      * @throws IOException
  63.      * @throws InterruptedException
  64.      */
  65.     public boolean nextKeyValue() throws IOException, InterruptedException {
  66.         if(flag){
  67.             //读取文件进入key和value
  68.             String path = fs.getPath().toString();
  69.             key.set(path);
  71.             //文件是一次性读完,bytes的长度不能为普遍的1024,当然这么写会涉及到大文件的问题,不做讨论.
  72.             byte[] bytes = new byte[(int) fs.getLength()];
  73.             fis.read(bytes);
  74.             value.set(bytes,0,bytes.length);
  76.             //重新标记
  77.             flag = false;
  79.             return  true;
  80.         }
  81.         return false;
  82.     }
  84.     /**
  85.      * 获取当前读到的key
  86.      * @return
  87.      * @throws IOException
  88.      * @throws InterruptedException
  89.      */
  90.     public Text getCurrentKey() throws IOException, InterruptedException {
  91.         return this.key;
  92.     }
  94.     /**
  95.      * 获取当前读到的value
  96.      * @return
  97.      * @throws IOException
  98.      * @throws InterruptedException
  99.      */
  100.     public BytesWritable getCurrentValue() throws IOException, InterruptedException {
  101.         return this.value;
  102.     }
  104.     /**
  105.      * 获取当前读取的进度
  106.      * @return
  107.      * @throws IOException
  108.      * @throws InterruptedException
  109.      */
  110.     public float getProgress() throws IOException, InterruptedException {
  111.         //文件一次读完,只有0和1的进度,根据flag来判断
  112.         return flag ? 0f : 1f;
  113.     }
  115.     /**
  116.      * 关闭资源
  117.      * @throws IOException
  118.      */
  119.     public void close() throws IOException {
  120.         IOUtils.closeStream(fis);
  121.     }
  122. }

2) 自定义Inputformat

  1. package com.nty.inputformat;
  3. import org.apache.hadoop.fs.Path;
  4. import org.apache.hadoop.io.BytesWritable;
  5. import org.apache.hadoop.io.Text;
  6. import org.apache.hadoop.mapreduce.InputSplit;
  7. import org.apache.hadoop.mapreduce.JobContext;
  8. import org.apache.hadoop.mapreduce.RecordReader;
  9. import org.apache.hadoop.mapreduce.TaskAttemptContext;
  10. import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
  12. import java.io.IOException;
  14. /**
  15.  * author nty
  16.  * date time 2018-12-11 9:09
  17.  */
  18. //需求中,key为文件路径+名称,所以key类型为Text,value为文件内容,用BytesWritable
  19. public class CustomInputFormat extends FileInputFormat<Text, BytesWritable> {
  21.     //最后输出的value为一个文件,所让文件不能被切分,返回false
  22.     @Override
  23.     protected boolean isSplitable(JobContext context, Path filename) {
  24.         return false;
  25.     }
  27.     //返回自定义的 RecordReader
  28.     public RecordReader<Text, BytesWritable> createRecordReader(InputSplit split, TaskAttemptContext context) throws IOException, InterruptedException {
  29.         return new CustomRecordReader();
  30.     }
  31. }

3) 编写Mapper类

  1. package com.nty.inputformat;
  3. import org.apache.hadoop.io.BytesWritable;
  4. import org.apache.hadoop.io.Text;
  5. import org.apache.hadoop.mapreduce.Mapper;
  7. import java.io.IOException;
  9. /**
  10.  * author nty
  11.  * date time 2018-12-11 9:10
  12.  */
  13. public class CustomMapper extends Mapper<Text, BytesWritable, Text, BytesWritable> {
  14.     @Override
  15.     protected void map(Text key, BytesWritable value, Context context) throws IOException, InterruptedException {
  16.         context.write(key,value);
  17.     }
  18. }

4) 编写Reducer类

  1. package com.nty.inputformat;
  3. import org.apache.hadoop.io.BytesWritable;
  4. import org.apache.hadoop.io.Text;
  5. import org.apache.hadoop.mapreduce.Reducer;
  7. import java.io.IOException;
  9. /**
  10.  * author nty
  11.  * date time 2018-12-11 9:10
  12.  */
  13. public class CustomReducer extends Reducer<Text, BytesWritable, Text, BytesWritable> {
  14.     @Override
  15.     protected void reduce(Text key, Iterable<BytesWritable> values, Context context) throws IOException, InterruptedException {
  16.         for (BytesWritable value : values) {
  17.             context.write(key, value);
  18.         }
  19.     }
  20. }

5) 编写Driver类

  1. package com.nty.inputformat;
  3. import org.apache.hadoop.conf.Configuration;
  4. import org.apache.hadoop.fs.Path;
  5. import org.apache.hadoop.io.BytesWritable;
  6. import org.apache.hadoop.io.Text;
  7. import org.apache.hadoop.mapreduce.Job;
  8. import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
  9. import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
  10. import org.apache.hadoop.mapreduce.lib.output.SequenceFileOutputFormat;
  12. /**
  13.  * author nty
  14.  * date time 2018-12-11 9:10
  15.  */
  16. public class CustomDriver {
  18.     public static void main(String[] args) throws  Exception{
  19.         //获取job
  20.         Configuration configuration = new Configuration();
  21.         Job job = Job.getInstance(configuration);
  23.         //设置类
  24.         job.setJarByClass(CustomDriver.class);
  25.         //设置input和output
  26.         job.setInputFormatClass(CustomInputFormat.class);
  27.         job.setOutputFormatClass(SequenceFileOutputFormat.class);
  29.         //设置Mapper和Reducer
  30.         job.setMapperClass(CustomMapper.class);
  31.         job.setReducerClass(CustomReducer.class);
  33.         //设置Mapper和Reducer的输入输出
  34.         job.setMapOutputKeyClass(Text.class);
  35.         job.setMapOutputValueClass(BytesWritable.class);
  37.         job.setOutputKeyClass(Text.class);
  38.         job.setOutputValueClass(BytesWritable.class);
  40.         //设置文件路径
  41.         FileInputFormat.setInputPaths(job, new Path("d:\\Hadoop_test"));
  42.         FileOutputFormat.setOutputPath(job, new Path("d:\\Hadoop_test_out"));
  43.         //提交
  44.         boolean b = job.waitForCompletion(true);
  46.         System.exit(b ? 0 : 1);
  48.     }
  49. }

 

 友情链接：直通硅谷 　点职佳 　北美留学生论坛

Hadoop热门文章

  • Hive中的去重 ：distinct,group by与ROW_Number()窗口函数
  • HBASE的安装过程及运行HBASE程序的需要配置的内容
  • hadoop--Unable to load native-hadoop library for your platform解决方法
  • Hive sampling 语法之TABLESAMPLE用法理解
  • idea项目运行时报错提示Could not locate executable null\bin\winutils.exe in the Hadoop binaries
  • hadoop 无法访问50070端口问题解决
  • Hadoop运行时遇到java.io.FileNotFoundException错误的解决方法
  • KUDU数据导入尝试一：TextFile数据导入Hive,Hive数据导入KUDU
  • Hive安装
  • Hadoop 学习笔记 生态

    • Hadoop推荐文章

  • 基于Hive的大数据分析系统
  • HBase 在统一内容平台业务的优化实践
  • hive窗口分析函数使用详解系列一
  • 两台实体机器4个虚拟机节点的Hadoop集群搭建（Ubuntu版）
  • 在阿里云和腾讯云的轻量应用服务器上搭建Hadoop集群
  • Haproxy搭建 Web 群集实现负载均衡
  • HBase Compaction 原理与线上调优实践
  • Hive执行计划之hive依赖及权限查询和常见使用场景
  • 工良出品：包教会，Hadoop、Hive 搭建部署简易教程
  • Centos7系统编译Hadoop3.3.4

    • 本站QQ群：前端 618073944 | Java 606181507 | Python 626812652 | C/C++ 612253063 | 微信 634508462 | 苹果 692586424 | C#/.net 182808419 | PHP 305140648 | 运维 608723728

    W3xue 的所有内容仅供测试，对任何法律问题及风险不承担任何责任。通过使用本站内容随之而来的风险与本站无关。
    关于我们  |  意见建议  |  捐助我们  |  报错有奖  |  广告合作、友情链接（目前9元/月）请联系QQ：27243702 沸活量
    皖ICP备17017327号-2　皖公网安备34020702000426号