MapReduce的常见输入格式之CombineTextInputFormat

青旅半醒 2022-11-14 14:52 46阅读 0赞

## CombineTextInputFormat ##

框架默认的 TextInputFormat 切片机制是对任务按文件规划切片，不管文件多小，都会  
是一个单独的切片，都会交给一个 MapTask，这样如果有大量小文件，就会产生大量的  
MapTask，处理效率极其低下。  
1 ） 应用场景：  
CombineTextInputFormat 用于小文件过多的场景，它可以将多个小文件从逻辑上规划到  
一个切片中，这样，多个小文件就可以交给一个 MapTask 处理。  
2 ） 虚拟存储 切片 最大值设置  
CombineTextInputFormat.setMaxInputSplitSize(job, 4194304);// 4m  
注意：虚拟存储切片最大值设置最好根据实际的小文件大小情况来设置具体的值。

作用： 改变了传统的切片方式，将多个小文件，划分到一个切片中，适合小文件过多的场景。

假设有4个文件a.text,b.text,c.text,d.text .

![20210330102455841.png][]

需求：将上面4个文件合并成一个切片统一处理

创建 WordCountMapper

>     /**
>      * KEYIN, map阶段输入的key的类型：LongWritable
>      * VALUEIN,map阶段输入value类型：Text
>      * KEYOUT,map阶段输出的Key类型：Text
>      * VALUEOUT,map阶段输出的value类型：IntWritable
>      * @author sun
>      */
>     public class WordCountMapper extends Mapper<LongWritable, Text, Text, IntWritable> {
>         private Text outK = new Text();
>         private IntWritable outV = new IntWritable(1);
>     
>         @Override
>         protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
>     
>             // 1 获取一行
>             String line = value.toString();
>     
>             // 2 切割
>             String[] words = line.split(" ");
>     
>             // 3 循环写出
>             for (String word : words) {
>                 // 封装outk
>                 outK.set(word);
>     
>                 // 写出
>                 context.write(outK, outV);
>             }
>         }
>     }

创建WordCountReducer

>     import org.apache.hadoop.io.IntWritable;
>     import org.apache.hadoop.io.Text;
>     import org.apache.hadoop.mapreduce.Reducer;
>     
>     import java.io.IOException;
>     
>     /**
>      * KEYIN, reduce阶段输入的key的类型：Text
>      * VALUEIN,reduce阶段输入value类型：IntWritable
>      * KEYOUT,reduce阶段输出的Key类型：Text
>      * VALUEOUT,reduce阶段输出的value类型：IntWritable
>      * @author sun
>      */
>     public class WordCountReducer extends Reducer<Text, IntWritable,Text,IntWritable> {
>         private IntWritable outV = new IntWritable();
>     
>         @Override
>         protected void reduce(Text key, Iterable<IntWritable> values, Context context) throws IOException, InterruptedException {
>     
>             int sum = 0;
>             // 累加
>             for (IntWritable value : values) {
>                 sum += value.get();
>             }
>     
>             outV.set(sum);
>     
>             // 写出
>             context.write(key,outV);
>         }
>     }

创建WordCountDriver，设置setInputFormatClass 为CombineTextInputFormat.class

// 如果不设置InputFormat，它默认用的是TextInputFormat.class,切片个数为 4

job.setInputFormatClass(CombineTextInputFormat.class);

>     import org.apache.hadoop.conf.Configuration;
>     import org.apache.hadoop.fs.Path;
>     import org.apache.hadoop.io.IntWritable;
>     import org.apache.hadoop.io.Text;
>     import org.apache.hadoop.mapreduce.Job;
>     import org.apache.hadoop.mapreduce.lib.input.CombineTextInputFormat;
>     import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
>     import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
>     
>     import java.io.IOException;
>     
>     /**
>      *  WordCountDriver
>      * @author sun
>      */
>     public class WordCountDriver {
>     
>         public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException {
>     
>             // 1 获取job
>             Configuration conf = new Configuration();
>             Job job = Job.getInstance(conf);
>     
>             // 2 设置jar包路径
>             job.setJarByClass(WordCountDriver.class);
>     
>             // 3 关联mapper和reducer
>             job.setMapperClass(WordCountMapper.class);
>             job.setReducerClass(WordCountReducer.class);
>     
>             // 4 设置map输出的kv类型
>             job.setMapOutputKeyClass(Text.class);
>             job.setMapOutputValueClass(IntWritable.class);
>     
>             // 5 设置最终输出的kV类型
>             job.setOutputKeyClass(Text.class);
>             job.setOutputValueClass(IntWritable.class);
>     
>             // 如果不设置InputFormat，它默认用的是TextInputFormat.class
>             job.setInputFormatClass(CombineTextInputFormat.class);
>     
>     
>            //虚拟存储切片最大值设置20m
>             CombineTextInputFormat.setMaxInputSplitSize(job, 20971520);
>     
>     
>             // 6 设置输入路径和输出路径
>             FileInputFormat.setInputPaths(job, new Path("D:\\input\\inputcombinetextinputformat"));
>             FileOutputFormat.setOutputPath(job, new Path("D:\\hadoop\\outputCombine3"));
>     
>             // 7 提交job
>             boolean result = job.waitForCompletion(true);
>     
>             System.exit(result ? 0 : 1);
>         }
>     }

[20210330102455841.png]: /images/20221022/c2a11da38cc34faf93d7a46ade149d60.png