小文件问题

image.png

需求： 10个小文件，统计里面单词出现的情况

image.png

SequenceFile

image.png
可以理解为把很多小文件压缩为一个大压缩包

  1. /**
  2.  * 小文件解决方案之SequenceFile
  3.  */
  4. public class SmallFileSeq {
  6.     public static void main(String[] args) throws Exception {
  7.         //生成SequenceFile文件
  8.         write("D:\\smallFile", "/seqFile");
  9.         //读取SequenceFile文件
  10.         read("/seqFile");
  11.     }
  14.     /**
  15.      * 生成SequenceFile文件
  16.      *
  17.      * @param inputDir   输入目录-windows目录
  18.      * @param outputFile 输出文件-hdfs文件
  19.      * @throws Exception
  20.      */
  21.     private static void write(String inputDir, String outputFile) throws Exception {
  22.         //创建一个配置对象
  23.         Configuration conf = new Configuration();
  24.         //指定HDFS的地址
  25.         conf.set("fs.defaultFS", "hdfs://bigdata1:9000");
  27.         //获取操作HDFS的对象
  28.         FileSystem fileSystem = FileSystem.get(conf);
  30.         //删除HDFS上的输出文件
  31.         fileSystem.delete(new Path(outputFile), true);
  33.         //构造opts数组，有三个元素
  34.         /*
  35.         第一个是输出路径【文件】
  36.         第二个是key的类型
  37.         第三个是value的类型
  38.          */
  39.         SequenceFile.Writer.Option[] opts = new SequenceFile.Writer.Option[]{
  40.                 SequenceFile.Writer.file(new Path(outputFile)),
  41.                 SequenceFile.Writer.keyClass(Text.class),
  42.                 SequenceFile.Writer.valueClass(Text.class)
  43.         };
  44.         //创建了一个writer实例
  45.         SequenceFile.Writer writer = SequenceFile.createWriter(conf, opts);
  47.         //指定需要压缩的文件的目录
  48.         File inputDirPath = new File(inputDir);
  49.         if (inputDirPath.isDirectory()) {
  50.             //获取目录中的文件
  51.             File[] files = inputDirPath.listFiles();
  52.             //迭代文件
  53.             for (File file : files) {
  54.                 //获取文件的全部内容
  55.                 String content = FileUtils.readFileToString(file, "UTF-8");
  56.                 //获取文件名
  57.                 String fileName = file.getName();
  58.                 Text key = new Text(fileName);
  59.                 Text value = new Text(content);
  60.                 //向SequenceFile中写入数据
  61.                 writer.append(key, value);
  62.             }
  63.         }
  64.         writer.close();
  65.     }
  67.     /**
  68.      * 读取SequenceFile文件
  69.      *
  70.      * @param inputFile SequenceFile文件路径
  71.      * @throws Exception
  72.      */
  73.     private static void read(String inputFile) throws Exception {
  74.         //创建一个配置对象
  75.         Configuration conf = new Configuration();
  76.         //指定HDFS的地址
  77.         conf.set("fs.defaultFS", "hdfs://bigdata1:9000");
  78.         //创建阅读器
  79.         SequenceFile.Reader reader = new SequenceFile.Reader(conf, SequenceFile.Reader.file(new Path(inputFile)));
  80.         Text key = new Text();
  81.         Text value = new Text();
  82.         //循环读取数据
  83.         while (reader.next(key, value)) {
  84.             //输出文件名称
  85.             System.out.print("文件名：" + key.toString() + ",");
  86.             //输出文件内容
  87.             System.out.println("文件内容：" + value.toString() + "");
  88.         }
  89.         reader.close();
  90.     }
  92. }

运行结果：
image.png

MapFile

image.png

  1. /**
  2.  * 小文件解决方案之MapFile
  3.  */
  4. public class SmallFileMap {
  6.     public static void main(String[] args) throws Exception{
  7.         //生成MapFile文件
  8.         write("D:\\smallFile","/mapFile");
  9.         //读取MapFile文件
  10.         read("/mapFile");
  11.     }
  13.     /**
  14.      * 生成MapFile文件
  15.      * @param inputDir 输入目录-windows目录
  16.      * @param outputDir 输出目录-hdfs目录
  17.      * @throws Exception
  18.      */
  19.     private static void write(String inputDir,String outputDir) throws Exception{
  20.         //创建一个配置对象
  21.         Configuration conf = new Configuration();
  22.         //指定HDFS的地址
  23.         conf.set("fs.defaultFS","hdfs://bigdata1:9000");
  25.         //获取操作HDFD的对象
  26.         FileSystem fileSystem = FileSystem.get(conf);
  28.         //删除HDFS上的输出文件
  29.         fileSystem.delete(new Path(outputDir),true);
  31.         //构造opts数组，有两个元素
  32.         /*
  33.         第一个是key的类型
  34.         第二个是value的类型
  35.          */
  36.         SequenceFile.Writer.Option[] opts = new SequenceFile.Writer.Option[]{
  37.                 MapFile.Writer.keyClass(Text.class),
  38.                 MapFile.Writer.valueClass(Text.class)
  39.         };
  40.         //创建了一个writer实例
  41.         MapFile.Writer writer = new MapFile.Writer(conf, new Path(outputDir), opts);
  43.         //指定需要压缩的文件的目录
  44.         File inputDirPath = new File(inputDir);
  45.         if(inputDirPath.isDirectory()){
  46.             //获取目录中的文件
  47.             File[] files = inputDirPath.listFiles();
  48.             //迭代文件
  49.             for (File file: files) {
  50.                 //获取文件的全部内容
  51.                 String content = FileUtils.readFileToString(file, "UTF-8");
  52.                 //获取文件名
  53.                 String fileName = file.getName();
  54.                 Text key = new Text(fileName);
  55.                 Text value = new Text(content);
  56.                 //向SequenceFile中写入数据
  57.                 writer.append(key,value);
  58.             }
  59.         }
  60.         writer.close();
  61.     }
  63.     /**
  64.      * 读取MapFile文件
  65.      * @param inputDir MapFile文件路径
  66.      * @throws Exception
  67.      */
  68.     private static void read(String inputDir)throws Exception{
  69.         //创建一个配置对象
  70.         Configuration conf = new Configuration();
  71.         //指定HDFS的地址
  72.         conf.set("fs.defaultFS","hdfs://bigdata1:9000");
  73.         //创建阅读器
  74.         MapFile.Reader reader = new MapFile.Reader(new Path(inputDir),conf);
  75.         Text key = new Text();
  76.         Text value = new Text();
  77.         //循环读取数据
  78.         while(reader.next(key,value)){
  79.             //输出文件名称
  80.             System.out.print("文件名："+key.toString()+",");
  81.             //输出文件内容
  82.             System.out.println("文件内容："+value.toString()+"");
  83.         }
  84.         reader.close();
  85.     }
  87. }

运行结果：
image.png

计算生成的SequenceFile

  1. /**
  2.  * 需求：读取SequenceFile文件
  3.  */
  4. @Slf4j
  5. public class WordCountJobSeq {
  6.     /**
  7.      * Map阶段
  8.      */
  9.     public static class MyMapper extends Mapper<Text, Text, Text, LongWritable> {
  10.         /**
  11.          * 需要实现map函数
  12.          * 这个map函数就是可以接收<k1,v1>，产生<k2，v2>
  13.          *
  14.          * @param k1      这里的k1不是偏移量，而是SequenceFile的文件名
  15.          * @param v1      文件内容
  16.          * @param context
  17.          * @throws IOException
  18.          * @throws InterruptedException
  19.          */
  20.         @Override
  21.         protected void map(Text k1, Text v1, Context context) throws IOException, InterruptedException {
  22.             log.info("<k1,v1>=<{},{}>", k1.toString(), v1.toString());
  24.             //对获取到的每一行数据进行切割，把单词切割出来
  25.             String[] words = v1.toString().split(" ");
  26.             //迭代切割出来的单词数据
  27.             for (String word : words) {
  28.                 //把迭代出来的单词封装成<k2,v2>的形式
  29.                 Text k2 = new Text(word);
  30.                 LongWritable v2 = new LongWritable(1L);
  31.                 //把<k2,v2>写出去
  32.                 context.write(k2, v2);
  33.             }
  34.         }
  35.     }
  38.     /**
  39.      * Reduce阶段
  40.      */
  41.     public static class MyReducer extends Reducer<Text, LongWritable, Text, LongWritable> {
  42.         /**
  43.          * 针对<k2,{v2...}>的数据进行累加求和，并且最终把数据转化为k3,v3写出去
  44.          *
  45.          * @param k2
  46.          * @param v2s
  47.          * @param context
  48.          * @throws IOException
  49.          * @throws InterruptedException
  50.          */
  51.         @Override
  52.         protected void reduce(Text k2, Iterable<LongWritable> v2s, Context context) throws IOException, InterruptedException {
  53.             //创建一个sum变量，保存v2s的和
  54.             long sum = 0L;
  55.             //对v2s中的数据进行累加求和
  56.             for (LongWritable v2 : v2s) {
  57.                 log.info("<k2,v2>=<{},{}>", k2.toString(), v2.get());
  58.                 sum += v2.get();
  59.             }
  61.             //组装k3,v3
  62.             Text k3 = k2;
  63.             LongWritable v3 = new LongWritable(sum);
  65.             log.info("<k3,v3>=<{},{}>", k3.toString(), v3.get());
  66.             // 把结果写出去
  67.             context.write(k3, v3);
  68.         }
  69.     }
  71.     /**
  72.      * 组装Job=Map+Reduce
  73.      */
  74.     public static void main(String[] args) {
  75.         try {
  76.             if (args.length != 2) {
  77.                 //如果传递的参数不够，程序直接退出
  78.                 System.exit(100);
  79.             }
  81.             //指定Job需要的配置参数
  82.             Configuration conf = new Configuration();
  83.             //创建一个Job
  84.             Job job = Job.getInstance(conf);
  86.             //注意了：这一行必须设置，否则在集群中执行的时候是找不到WordCountJob这个类的
  87.             job.setJarByClass(WordCountJobSeq.class);
  89.             //指定输入路径（可以是文件，也可以是目录）
  90.             FileInputFormat.setInputPaths(job, new Path(args[0]));
  91.             //指定输出路径(只能指定一个不存在的目录)
  92.             FileOutputFormat.setOutputPath(job, new Path(args[1]));
  94.             //指定map相关的代码
  95.             job.setMapperClass(MyMapper.class);
  96.             //指定k2的类型
  97.             job.setMapOutputKeyClass(Text.class);
  98.             //指定v2的类型
  99.             job.setMapOutputValueClass(LongWritable.class);
  101.             //设置输入数据处理类
  102.             job.setInputFormatClass(SequenceFileInputFormat.class);
  105.             //指定reduce相关的代码
  106.             job.setReducerClass(MyReducer.class);
  107.             //指定k3的类型
  108.             job.setOutputKeyClass(Text.class);
  109.             //指定v3的类型
  110.             job.setOutputValueClass(LongWritable.class);
  112.             //提交job
  113.             job.waitForCompletion(true);
  114.         } catch (Exception e) {
  115.             e.printStackTrace();
  116.         }
  118.     }
  120. }

计算之前生成的seqFile

hadoop jar db_hadoop-1.0-SNAPSHOT-jar-with-dependencies.jar com.imooc.mr.WordCountJobSeq /seqFile /out10

查看计算结果
image.png
查看日志
Map阶段：
image.png
Reduce阶段
image.png

总结

1、将一堆小文件合并为一个SequenceFile
2、使用MapReduce计算SequenceFile

附：一个小文件有多行的情况时，可能需要修改

image.png
image.png
image.png