通过SequenceFile实现合并小文件(调优技能)

痛定思痛。 2021-09-23 14:30 215阅读 0赞

### 文章目录 ###

*  0x00 文章内容
 *  0x01 通过SequenceFile合并小文件
 *   *   *   *   *  1. 准备工作
                 *  2. 完整代码
 *  0x02 检验结果
 *   *   *   *   *  1. 启动HDFS和YARN
                 *  2. 执行作业
                 *  3. 查看执行结果
 *  0xFF 总结

# 0x00 文章内容 #

1.  通过SequenceFile合并小文件
2.  检验结果

说明：Hadoop集群中，元数据是交由NameNode来管理的，每个小文件就是一个split，会有自己相对应的元数据，如果小文件很多，则会对内存以及NameNode很大的压力，所以可以通过合并小文件的方式来进行优化。合并小文件其实可以有两种方式：一种是通过Sequence格式转换文件来合并，另一种是通过CombineFileInputFormat来实现。

此处选择SequeceFile类型是因为此格式为二进制格式，而且是key-value类型，我们在合并小文件的时候，可以利用此特性，将每个小文件的名称做为key，将每个小文件里面的内容做为value。

# 0x01 通过SequenceFile合并小文件 #

##### 1. 准备工作 #####

a. 我的HDFS上有四个文件：

[hadoop-sny@master ~]$ hadoop fs -ls /files/
    Found 4 items
    -rw-r--r--   1 hadoop-sny supergroup         39 2019-04-18 21:20 /files/put.txt
    -rw-r--r--   1 hadoop-sny supergroup         50 2019-12-30 17:12 /files/small1.txt
    -rw-r--r--   1 hadoop-sny supergroup         31 2019-12-30 17:10 /files/small2.txt
    -rw-r--r--   1 hadoop-sny supergroup         49 2019-12-30 17:11 /files/small3.txt

内容对应如下，其实内容可以随意：

shao nai yi
    nai nai yi yi
    shao nai nai

hello hi hi hadoop
    spark kafka shao
    nai yi nai yi

hello 1
    hi 1
    shao 3
    nai 1
    yi 3

guangdong 300
    hebei 200
    beijing 198
    tianjing 209

b. 除了在Linux上创建然后上传外，还可以直接以流的方式输入进去，如`small1.txt`：

`hadoop fs -put - /files/small1.txt`

输入完后，按`ctrl` \+ `D` 结束输入。

##### 2. 完整代码 #####

a. `SmallFilesToSequenceFileConverter`完整代码

package com.shaonaiyi.hadoop.filetype.smallfiles;
    
    import com.shaonaiyi.hadoop.utils.FileUtils;
    import org.apache.hadoop.conf.Configuration;
    import org.apache.hadoop.fs.Path;
    import org.apache.hadoop.io.BytesWritable;
    import org.apache.hadoop.io.NullWritable;
    import org.apache.hadoop.io.Text;
    import org.apache.hadoop.mapreduce.InputSplit;
    import org.apache.hadoop.mapreduce.Job;
    import org.apache.hadoop.mapreduce.Mapper;
    import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
    import org.apache.hadoop.mapreduce.lib.input.FileSplit;
    import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
    import org.apache.hadoop.mapreduce.lib.output.SequenceFileOutputFormat;
    
    import java.io.IOException;
    /** * @Author shaonaiyi@163.com * @Date 2019/12/30 16:29 * @Description 通过SequenceFile合并小文件 */
    public class SmallFilesToSequenceFileConverter { 
    
        static class SequenceFileMapper extends Mapper<NullWritable, BytesWritable, Text, BytesWritable> { 
            private Text fileNameKey;
    
            @Override
            protected void setup(Context context) { 
                InputSplit split = context.getInputSplit();
                Path path = ((FileSplit) split).getPath();
                fileNameKey = new Text(path.toString());
            }
    
            @Override
            protected void map(NullWritable key, BytesWritable value, Context context) throws IOException, InterruptedException { 
                context.write(fileNameKey, value);
            }
        }
    
        public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException { 
            Job job = Job.getInstance(new Configuration(), "SmallFilesToSequenceFileConverter");
    
            job.setJarByClass(SmallFilesToSequenceFileConverter.class);
    
            job.setInputFormatClass(WholeFileInputFormat.class);
    
            job.setOutputKeyClass(Text.class);
            job.setOutputValueClass(BytesWritable.class);
            job.setOutputFormatClass(SequenceFileOutputFormat.class);
    
            job.setMapperClass(SequenceFileMapper.class);
    
            FileInputFormat.addInputPath(job, new Path(args[0]));
    
            String outputPath = args[1];
            FileUtils.deleteFileIfExists(outputPath);
            FileOutputFormat.setOutputPath(job, new Path(outputPath));
            System.exit(job.waitForCompletion(true) ? 0 : 1);
        }
    
    }

b. `WholeFileInputFormat`完整代码

package com.shaonaiyi.hadoop.filetype.smallfiles;
    
    import org.apache.hadoop.fs.Path;
    import org.apache.hadoop.io.BytesWritable;
    import org.apache.hadoop.io.NullWritable;
    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;
    
    /** * @Author shaonaiyi@163.com * @Date 2019/12/30 16:34 * @Description 实现WholeFileInputFormat类 */
    public class WholeFileInputFormat extends FileInputFormat<NullWritable, BytesWritable> { 
    
        @Override
        protected boolean isSplitable(JobContext context, Path filename) { 
            return false;
        }
    
        @Override
        public RecordReader<NullWritable, BytesWritable> createRecordReader(InputSplit inputSplit, TaskAttemptContext taskAttemptContext) throws IOException, InterruptedException { 
            WholeFileRecordReader reader = new WholeFileRecordReader();
            reader.initialize(inputSplit, taskAttemptContext);
            return reader;
        }
    }

c. `WholeFileRecordReader`完整代码

package com.shaonaiyi.hadoop.filetype.smallfiles;
    
    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.NullWritable;
    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;
    
    /** * @Author shaonaiyi@163.com * @Date 2019/12/30 16:35 * @Description 实现WholeFileRecordReader类 */
    public class WholeFileRecordReader extends RecordReader<NullWritable, BytesWritable> { 
    
        private FileSplit fileSplit;
        private Configuration configuration;
        private BytesWritable value = new BytesWritable();
        private boolean processed = false;
    
        @Override
        public void initialize(InputSplit inputSplit, TaskAttemptContext taskAttemptContext) throws IOException, InterruptedException { 
            this.fileSplit = (FileSplit)inputSplit;
            this.configuration = taskAttemptContext.getConfiguration();
        }
    
        @Override
        public boolean nextKeyValue() throws IOException, InterruptedException { 
            if (!processed) { 
                byte[] contents = new byte[(int)fileSplit.getLength()];
                Path file = fileSplit.getPath();
                FileSystem fs = file.getFileSystem(configuration);
                FSDataInputStream in = null;
                try { 
                    in = fs.open(file);
                    IOUtils.readFully(in, contents, 0, contents.length);
                    value.set(contents, 0, contents.length);
                } finally { 
                    IOUtils.closeStream(in);
                }
                processed = true;
                return true;
            }
            return false;
        }
    
        @Override
        public NullWritable getCurrentKey() throws IOException, InterruptedException { 
            return NullWritable.get();
        }
    
        @Override
        public BytesWritable getCurrentValue() throws IOException, InterruptedException { 
            return value;
        }
    
        @Override
        public float getProgress() throws IOException, InterruptedException { 
            return processed ? 1.0f : 0.0f;
        }
    
        @Override
        public void close() throws IOException { 
    
        }
    }

# 0x02 检验结果 #

##### 1. 启动HDFS和YARN #####

`start-dfs.sh`  
`start-yarn.sh`

##### 2. 执行作业 #####

a. 打包并上传到master上执行，需要传入两个参数

yarn jar ~/jar/hadoop-learning-1.0.jar com.shaonaiyi.hadoop.filetype.smallfiles.SmallFilesToSequenceFileConverter /files /output

##### 3. 查看执行结果 #####

a. 生成了一份文件  
![在这里插入图片描述][20191230172658115.png]  
b. 查看到里面的内容如下，但内容很难看  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3NoYW9jazIwMTg_size_16_color_FFFFFF_t_70]  
c. 用text查看文件内容，可看到key为文件名，value为二进制的里面的内容。![在这里插入图片描述][20191230172741586.png]

# 0xFF 总结 #

1.  Input的路径有4个文件，默认会启动4个mapTask，其实我们可以通过`CombineTextInputFormat`设置成只启动一个：

job.setInputFormatClass(CombineTextInputFormat.class);

具体操作请参考教程：[通过CombineTextInputFormat实现合并小文件(调优技能)][CombineTextInputFormat]

--------------------

**福利：**  
[邵奈一的技术博客导航][Link 6]  
[邵奈一][CSDN] 原创不易，如转载请标明出处。

--------------------

[20191230172658115.png]: /images/20210923/651b8523ae6d47d0a945883940a16f54.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3NoYW9jazIwMTg_size_16_color_FFFFFF_t_70]: /images/20210923/b178845346b94b26abe0b0b31bedb76c.png
[20191230172741586.png]: /images/20210923/d48d0f1637e14423998b0ffab0fb4c40.png
[CombineTextInputFormat]: https://blog.csdn.net/shaock2018/article/details/103770117
[Link 1]: https://blog.csdn.net/shaock2018/article/details/91184254
[Link 2]: https://img-blog.csdnimg.cn/20190608000417972.jpeg?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3NoYW9jazIwMTg=,size_16,color_FFFFFF,t_70
[Link 3]: https://img-blog.csdnimg.cn/20190514084602748.png
[Link 4]: https://weibo.com/5680168183/
[CSDN]: https://blog.csdn.net/shaock2018
[Link 5]: https://www.jianshu.com/u/6f033c6a3b90
[Link 6]: https://blog.csdn.net/shaock2018/article/details/89349249