【Hadoop】Hadoop生态系列之InputForamt.class与OutputFormat.class分析

太过爱你忘了你带给我的痛 2023-10-05 10:52 55阅读 0赞

上一篇：Hadoop生态系列之MapReduce概述及MapReduce任务开发与发布

指路牌

    • InputForamt&OutputFormat
      • 整体设计
      • InputFormat.class
        • TextInputFormat.class
        • NLineInputFormat.class
        • KeyValueTextInputFormat.class
        • MultipleInputs.class
        • CombineFileInputFormat.class
        • DBInputFormat.class
      • OutputFormat.classs
        • TextoutputFormat.class
        • DBOutputFormat.class
        • JedisOutputFormat.class
        • 依赖解决
    • InputFormat/OuputFormat与Mapper和Reducer

InputForamt&OutputFormat

整体设计

\[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-8JglVn8m-1601378732933)(F:/BigData/hadoop生态系列课程/assets/image-20200929100753565.png)\]

InputFormat.class

该类是Hadoop提供的顶层抽象类，该类主要定制切片计算逻辑和切片数据的读取逻辑。

  1. public abstract class InputFormat<K, V> {
  3.     public InputFormat() {
  5.     }
  6.     //计算切片/ 数据拆分逻辑 区间
  7.     public abstract List<InputSplit> getSplits(JobContext var1) 
  8.         throws IOException, InterruptedException;
  9.     //实现逻辑区间的读取逻辑，将读取的数据传递给Mapper
  10.     public abstract RecordReader<K, V> createRecordReader(InputSplit var1,TaskAttemptContext var2) 
  11.         throws IOException, InterruptedException;
  12. }

在Hadoop的实现包中提供了InputFormat.class接口预实现主要有：

  • CompositeInputFormat.class - 主要实现在Map段实现大规模数据集的join
  • DBInputFormat - 主要提供针对RDBMS数据库的读取实现，主要针对Oracle和MySQL数据库。
  • FileInputFormat -主要针对分布式文件系统提供的预实现。
TextInputFormat.class

默认将文件按照128MB的大小为单位进行切割，切割的区间称为一个Split ，然后使用LineRecordReader对区间数据进行读取，该LineRecordReader会给Mapper提供每一行文本数据作为value，同时提供该value在文本行中的字节偏移量该偏移量是一个Long类型的参数，通常没有什么作用。

注意：默认FileInputFormat 的所有子类，在没有覆盖 getSplits方法的时候，默认计算的切片大小的区间（0，140.8MB]因为底层在计算文件切片的时候是通过(文件的大小/128MB > 1.1)?切分新块:不切分

代码参考：Hadoop生态系列之MapReduce概述及MapReduce任务开发与发布

NLineInputFormat.class

默认将文件按照N行对文件进行切割，切割的区间称为一个Split ，然后使用LineRecordReader对区间数据进行读取，该LineRecordReader会给Mapper提供每一行文本数据作为value，同时提供该value在文本行中的字节偏移量该偏移量是一个Long类型的参数，通常没有什么作用。

覆盖了FileInputFormat的getSplits,因此我们在使用NLineInputFormat的时候一般需要设置行数。

  1. NLineInputFormat.setNumLinesPerSplit(job,1000);
  3. public class URLCountApplication extends Configured implements Tool {
  5.     public int run(String[] strings) throws Exception {
  7.         //1.创建一个Job对象
  8.         Configuration conf = getConf();
  10.         Job job= Job.getInstance(conf,"URLCountApplication");
  12.         //2.告诉job数据格式
  13.         job.setInputFormatClass(NLineInputFormat.class);
  14.         NLineInputFormat.setNumLinesPerSplit(job,1000);
  15.         job.setOutputFormatClass(TextOutputFormat.class);
  17.         //3.设置数据路径
  18.         TextInputFormat.addInputPath(job,new Path("D:/data/click"));
  19.         //系统自动创建，如果在执行前存在，则放弃执行
  20.         TextOutputFormat.setOutputPath(job,new Path("D:/data/result"));
  22.         //4.设置处理逻辑
  23.         job.setMapperClass(URLMapper.class);
  24.         job.setReducerClass(URLReducer.class);
  26.         //5.设置输出的Key,value
  27.         job.setMapOutputKeyClass(Text.class);
  28.         job.setMapOutputValueClass(IntWritable.class);
  30.         job.setOutputKeyClass(Text.class);
  31.         job.setOutputValueClass(IntWritable.class);
  33.         //6.提交job
  34.         return job.waitForCompletion(true)?1:0;
  35.     }
  36.     public static void main(String[] args) throws Exception {
  38.         ToolRunner.run(new URLCountApplication(),args);
  39.     }
  40. }
KeyValueTextInputFormat.class

默认将文件按照128MB的大小为单位进行切割，切割的区间称为一个Split ，然后使用KeyValueLineRecordReader对区间数据进行读取，该KeyValueLineRecordReader会给Mapper提供key和value都是Text类型，该格式输入默认按照制表符\t进行分隔Key/Value如果没有正确拆分，会将整行作为key，value为null

  1. conf.set(KeyValueLineRecordReader.KEY_VALUE_SEPERATOR,",");
  3. public class URLCountApplication extends Configured implements Tool {
  5.     public int run(String[] strings) throws Exception {
  7.         //1.创建一个Job对象
  8.         Configuration conf = getConf();
  9.         conf.set(KeyValueLineRecordReader.KEY_VALUE_SEPERATOR,",");
  11.         Job job= Job.getInstance(conf,"AvgCostAplication");
  13.         //2.告诉job数据格式
  14.         job.setInputFormatClass(KeyValueTextInputFormat.class);
  15.         job.setOutputFormatClass(TextOutputFormat.class);
  17.         //3.设置数据路径
  18.         TextInputFormat.addInputPath(job,new Path("file:///D:/data/keyvalue"));
  19.         //系统自动创建，如果在执行前存在，则放弃执行
  20.         TextOutputFormat.setOutputPath(job,new Path("file:///D:/data/result"));
  22.         //4.设置处理逻辑
  23.         job.setMapperClass(AvgCostMapper.class);
  24.         job.setReducerClass(AvgCostReducer.class);
  26.         //5.设置输出的Key,value
  27.         job.setMapOutputKeyClass(Text.class);
  28.         job.setMapOutputValueClass(DoubleWritable.class);
  30.         job.setOutputKeyClass(Text.class);
  31.         job.setOutputValueClass(DoubleWritable.class);
  33.         //6.提交job
  34.         return job.waitForCompletion(true)?1:0;
  35.     }
  36.     public static void main(String[] args) throws Exception {
  38.         ToolRunner.run(new URLCountApplication(),args);
  39.     }
  40. }
MultipleInputs.class

这是一个复合的输入格式，主要适用于将多个不同格式的InputFormat组合使用，要求Map端的输出格式还必须保持一致。

  1. public class SumCostApplication extends Configured implements Tool {
  3.     public int run(String[] strings) throws Exception {
  5.         //1.创建一个Job对象
  6.         Configuration conf = getConf();
  7.         conf.set(KeyValueLineRecordReader.KEY_VALUE_SEPERATOR,",");
  8.         Job job= Job.getInstance(conf,"SumCostCountApplication");
  10.         //2.告诉job数据格式
  11.         job.setOutputFormatClass(TextOutputFormat.class);
  13.         //3.设置数据路径
  14.         TextOutputFormat.setOutputPath(job,new Path("file:///D:/data/result"));
  15.         //4.设置处理逻辑
  16.         MultipleInputs.addInputPath(job,new Path("file:///D:/data/mul/keyvalue"), KeyValueTextInputFormat.class,KeyVlaueCostMapper.class);
  17.         MultipleInputs.addInputPath(job,new Path("file:///D:/data/mul/text"), TextInputFormat.class,TextCostMapper.class);
  18.         job.setReducerClass(CostSumReducer.class);
  20.         //5.设置输出的Key,value
  21.         job.setMapOutputKeyClass(Text.class);
  22.         job.setMapOutputValueClass(DoubleWritable.class);
  24.         job.setOutputKeyClass(Text.class);
  25.         job.setOutputValueClass(DoubleWritable.class);
  27.         //6.提交job
  28.         return job.waitForCompletion(true)?1:0;
  29.     }
  30.     public static void main(String[] args) throws Exception {
  32.         ToolRunner.run(new SumCostApplication(),args);
  33.     }
  34. }
CombineFileInputFormat.class

上述的所有FileInputFormat都是以文件为单位计算文件切片，也就意味着如果计算的目录下有很多小文件，就会导致第一 阶段的Map任务过多。因此默认FileInputFormat对小文件处理不是太友好，因此Hadoop提供了CombineFileInputFormat格式类，该类专门用于处理小文件场景下的切片计算，会将多个小文件对应同一个切片。但是要求这些小文件的格式必须一致。我们可以使用CombineTextInputFormat该类和TextInputFormat用法一致，不同的是在于切片的计算上。

  1. public class URLCountApplication extends Configured implements Tool {
  3.     public int run(String[] strings) throws Exception {
  5.         //1.创建一个Job对象
  6.         Configuration conf = getConf();
  8.         Job job= Job.getInstance(conf,"URLCountApplication");
  10.         //2.告诉job数据格式
  11.         job.setInputFormatClass(CombineTextInputFormat.class);
  12.         job.setOutputFormatClass(TextOutputFormat.class);
  14.         //3.设置数据路径
  15.         CombineTextInputFormat.addInputPath(job,new Path("file:///D:/data/click"));
  16.         //系统自动创建，如果在执行前存在，则放弃执行
  17.         TextOutputFormat.setOutputPath(job,new Path("file:///D:/data/result"));
  19.         //4.设置处理逻辑
  20.         job.setMapperClass(URLMapper.class);
  21.         job.setReducerClass(URLReducer.class);
  23.         //5.设置输出的Key,value
  24.         job.setMapOutputKeyClass(Text.class);
  25.         job.setMapOutputValueClass(IntWritable.class);
  27.         job.setOutputKeyClass(Text.class);
  28.         job.setOutputValueClass(IntWritable.class);
  30.         //6.提交job
  31.         return job.waitForCompletion(true)?1:0;
  32.     }
  33.     public static void main(String[] args) throws Exception {
  35.         ToolRunner.run(new URLCountApplication(),args);
  36.     }
  37. }
DBInputFormat.class

主要负责读取RDBMS(关系型数据库)中的数据目前仅仅支持MySQL / Oracle数据库

  1. public class UserDBWritable implements DBWritable {
  3.     private Boolean sex;
  4.     private Double salary;
  5.     /**
  6.      * DBOutputFormat使用的
  7.      * @param statement
  8.      * @throws SQLException
  9.      */
  10.     public void write(PreparedStatement statement) throws SQLException {
  13.     }
  15.     public void readFields(ResultSet resultSet) throws SQLException {
  17.         this.sex=resultSet.getBoolean("sex");
  18.         this.salary=resultSet.getDouble("salary");
  19.     }
  21.     public Boolean getSex() {
  23.         return sex;
  24.     }
  26.     public void setSex(Boolean sex) {
  28.         this.sex = sex;
  29.     }
  31.     public Double getSalary() {
  33.         return salary;
  34.     }
  36.     public void setSalary(Double salary) {
  38.         this.salary = salary;
  39.     }
  40. }
  42. public class DBAvgSalaryApplication extends Configured implements Tool {
  44.     public int run(String[] strings) throws Exception {
  46.         //1.创建一个Job对象
  47.         Configuration conf = getConf();
  48.         //设置并行度
  49.         conf.setInt(MRJobConfig.NUM_MAPS,5);
  50.         DBConfiguration.configureDB(conf,"com.mysql.jdbc.Driver",
  51.                                     "jdbc:mysql://localhost:3306/test","root","123456");
  53.         Job job= Job.getInstance(conf,"DBAvgSalaryApplication");
  55.         //2.告诉job数据格式
  56.         job.setInputFormatClass(DBInputFormat.class);
  58.         String query="select sex,salary from t_user";
  59.         String countQuery="select count(*) from t_user";
  60.         DBInputFormat.setInput(job,UserDBWritable.class,query,countQuery);
  62.         job.setOutputFormatClass(TextOutputFormat.class);
  64.         //3.设置数据路径
  65.         //系统自动创建，如果在执行前存在，则放弃执行
  66.         TextOutputFormat.setOutputPath(job,new Path("D:/data/result"));
  68.         //4.设置处理逻辑
  69.         job.setMapperClass(UserAvgMapper.class);
  70.         job.setReducerClass(UserAvgReducer.class);
  72.         //5.设置输出的Key,value
  73.         job.setMapOutputKeyClass(BooleanWritable.class);
  74.         job.setMapOutputValueClass(DoubleWritable.class);
  76.         job.setOutputKeyClass(Text.class);
  77.         job.setOutputValueClass(DoubleWritable.class);
  79.         //6.提交job
  80.         return job.waitForCompletion(true)?1:0;
  81.     }
  82.     public static void main(String[] args) throws Exception {
  84.         ToolRunner.run(new DBAvgSalaryApplication(),args);
  85.     }
  86. }

OutputFormat.classs

该类是Hadoop提供的顶层抽象类，该类主要实现写的逻辑负责将Reduce端的输出写出到外围系统，同时也提供了输出检查（仅仅限于文件系统），负责返回Committer，确保系统能正常的输出。

  1. public abstract class OutputFormat<K, V> {
  4.   //创建RecordWriter
  5.   public abstract RecordWriter<K, V> 
  6.     getRecordWriter(TaskAttemptContext context
  7.                     ) throws IOException, InterruptedException;
  9.   //键查输出目录是否有效
  10.   public abstract void checkOutputSpecs(JobContext context
  11.                                         ) throws IOException, 
  12.                                                  InterruptedException;
  14.   //返回一个提交器
  15.   public abstract 
  16.   OutputCommitter getOutputCommitter(TaskAttemptContext context
  17.                                      ) throws IOException, InterruptedException;
  18. }
TextoutputFormat.class

将Reducer端的输出直接写入到文件系统中，其中在写入的时候会调用key、value的toString方法。

DBOutputFormat.class

将将Reducer端的输出直接写入到数据库系统中。

  1. public class URLCountDBWritable implements DBWritable {
  3.     private String category;
  4.     private Integer count;
  6.     public URLCountDBWritable(String category, Integer count) {
  8.         this.category = category;
  9.         this.count = count;
  10.     }
  12.     public URLCountDBWritable() {
  14.     }
  16.     public String getCategory() {
  18.         return category;
  19.     }
  21.     public void setCategory(String category) {
  23.         this.category = category;
  24.     }
  26.     public Integer getCount() {
  28.         return count;
  29.     }
  31.     public void setCount(Integer count) {
  33.         this.count = count;
  34.     }
  36.     public void write(PreparedStatement statement) throws SQLException {
  38.         statement.setString(1,category);
  39.         statement.setInt(2,count);
  40.     }
  42.     public void readFields(ResultSet resultSet) throws SQLException {
  45.     }
  46. }
  48. public class URLCountApplication extends Configured implements Tool {
  50.     public int run(String[] strings) throws Exception {
  52.         //1.创建一个Job对象
  53.         Configuration conf = getConf();
  54.         DBConfiguration.configureDB(conf,"com.mysql.jdbc.Driver",
  55.                 "jdbc:mysql://localhost:3306/test",
  56.                 "root","123456");
  57.         Job job= Job.getInstance(conf,"URLCountApplication");
  59.         //2.告诉job数据格式
  60.         job.setInputFormatClass(TextInputFormat.class);
  61.         job.setOutputFormatClass(DBOutputFormat.class);
  63.         //3.设置数据路径
  64.         TextInputFormat.addInputPath(job,new Path("file:///D:/data/click"));
  65.         DBOutputFormat.setOutput(job,"url_click","url_category","url_count");
  67.         //4.设置处理逻辑
  68.         job.setMapperClass(URLMapper.class);
  69.         job.setReducerClass(URLReducer.class);
  71.         //5.设置输出的Key,value
  72.         job.setMapOutputKeyClass(Text.class);
  73.         job.setMapOutputValueClass(IntWritable.class);
  75.         job.setOutputKeyClass(URLCountDBWritable.class);
  76.         job.setOutputValueClass(NullWritable.class);
  78.         //6.提交job
  79.         return job.waitForCompletion(true)?1:0;
  80.     }
  81.     public static void main(String[] args) throws Exception {
  83.         ToolRunner.run(new URLCountApplication(),args);
  84.     }
  85. }
JedisOutputFormat.class
  1. public class JedisOutputFormat extends OutputFormat<String,String> {
  3.     public final static String JEDIS_HOST="jedis.host";
  4.     public final static String JEDIS_PORT="jedis.port";
  6.     public static void setOutput(Job job, String host, Integer port) {
  8.         job.getConfiguration().set(JEDIS_HOST,host);
  9.         job.getConfiguration().setInt(JEDIS_PORT,port);
  10.     }
  12.     public RecordWriter<String, String> getRecordWriter(TaskAttemptContext context) throws IOException, InterruptedException {
  14.         Configuration config = context.getConfiguration();
  15.         String host=config.get(JEDIS_HOST);
  16.         Integer port=config.getInt(JEDIS_PORT,6379);
  17.         return new JedisRecordWriter(host,port);
  18.     }
  20.     public void checkOutputSpecs(JobContext context) throws IOException, InterruptedException {
  21.     }
  23.     public OutputCommitter getOutputCommitter(TaskAttemptContext context) throws IOException, InterruptedException {
  25.         return new FileOutputCommitter(FileOutputFormat.getOutputPath(context),
  26.                 context);
  27.     }
  28. }
  30. public class JedisRecordWriter  extends RecordWriter<String,String> {
  32.     private Jedis jedis=null;
  34.     public JedisRecordWriter(String host, Integer port) {
  36.         jedis=new Jedis(host,port);
  37.     }
  39.     public void write(String key, String value) throws IOException, InterruptedException {
  41.         jedis.set(key,value);
  42.     }
  44.     public void close(TaskAttemptContext context) throws IOException, InterruptedException {
  46.         jedis.close();
  47.     }
  48. }
  50. public class URLCountApplication extends Configured implements Tool {
  52.     public int run(String[] strings) throws Exception {
  54.         //1.创建一个Job对象
  55.         Configuration conf = getConf();
  57.         Job job= Job.getInstance(conf,"URLCountApplication");
  59.         //2.告诉job数据格式
  60.         job.setInputFormatClass(TextInputFormat.class);
  61.         job.setOutputFormatClass(JedisOutputFormat.class);
  63.         //3.设置数据路径
  64.         TextInputFormat.addInputPath(job,new Path("file:///D:/data/click"));
  65.         JedisOutputFormat.setOutput(job,"CentOS",6379);
  66.         //4.设置处理逻辑
  67.         job.setMapperClass(URLMapper.class);
  68.         job.setReducerClass(URLReducer.class);
  70.         //5.设置输出的Key,value
  71.         job.setMapOutputKeyClass(Text.class);
  72.         job.setMapOutputValueClass(IntWritable.class);
  74.         job.setOutputKeyClass(String.class);
  75.         job.setOutputValueClass(String.class);
  77.         //6.提交job
  78.         return job.waitForCompletion(true)?1:0;
  79.     }
  80.     public static void main(String[] args) throws Exception {
  82.         ToolRunner.run(new URLCountApplication(),args);
  83.     }
  84. }
依赖解决

  • 运行时依赖（Yarn Child依赖）

    • 方案1
      要求用户将依赖的jar包拷贝给所有的计算节点（NodeManager运行所在主机）

      1. [root@CentOS ~]# hadoop jar  xxx.jar 入口类  -libjars 依赖jar包1,依赖jar包2,....

    • 方案2

      1. [root@CentOS ~]# hdfs dfs -mkdir /libs
      2. [root@CentOS ~]# hdfs dfs -put mysql-connector-java-5.1.46.jar /libs
      4. conf.setStrings("tmpjars","/libs/xxx1.jar,/libs/xxx2.jar,...");
  • 提交时依赖（client ）

需要用户配置HADOOP_CLASSPATH环境变量(/root/.bashrc)，通常这种依赖发生在切片计算阶段。

  1. HADOOP_CLASSPATH=/root/mysql-connector-java-5.1.46.jar
  2. export HADOOP_CLASSPATH
  3. [root@CentOS ~]# source .bashrc 
  4. [root@CentOS ~]# hadoop classpath #查看hadoop的类路径
  5. /usr/hadoop-2.6.0/etc/hadoop:/usr/hadoop-2.6.0/share/hadoop/common/lib/*:/usr/hadoop-2.6.0/share/hadoop/common/*:/usr/hadoop-2.6.0/share/hadoop/hdfs:/usr/hadoop-2.6.0/share/hadoop/hdfs/lib/*:/usr/hadoop-2.6.0/share/hadoop/hdfs/*:/usr/hadoop-2.6.0/share/hadoop/yarn/lib/*:/usr/hadoop-2.6.0/share/hadoop/yarn/*:/usr/hadoop-2.6.0/share/hadoop/mapreduce/lib/*:/usr/hadoop-2.6.0/share/hadoop/mapreduce/*:`/root/mysql-connector-java-5.1.46.jar`:/usr/hadoop-2.6.0/contrib/capacity-scheduler/*.jar

InputFormat/OuputFormat与Mapper和Reducer

在这里插入图片描述

发表评论

表情：
评论列表 （有 0 条评论，55人围观）

还没有评论，来说两句吧...

相关阅读