一.MapReduce性能优化概述 1. MapReduce的应用场景
Hadoop包含了GFS的开源实现HDFS（Hadoop distributed file system)和MapReduce框架的开源实现。Hadoop得到了企业界及学术界关注，Yahoo、Facebook、Cloudera、Twitter、Intel、华为等诸多公司和技术团体对Hadoop给予了大力支持。Cloudera对Apache Hadoop及相关组件的版本兼容性进行了整合、性能优化、功能测试，推出了其企业版的开源Hadoop。Intel推出了高效、安全及易于管理的Hadoop企业版。Hadoop由于其开源性质，已成为目前研究、优化云计算框架的重要样本和基础。其中MapReduce框架很适合处理文档分析、倒排索引建立等类型的应用，然而在列存储、索引建立、连接计算、迭代计算、科学计算及调度算法方面性能需要进一步优化。
2. 优缺点与需求
1. 优点

1. Mapreduce易于编程
   它简单的实现一些接口，就可以完成一个分布式程序，这个程序可以分布到大量的廉价的pc机器上运行。也就是说你写一个分布式程序，跟写一个简单的串行程序是一模一样的。就是因为这个特性使的Mapreduce编程变得非常流行。
2. 良好的扩展性
   项目中当你的计算资源得不到满足的时候，你可以通过简单的通过增加机器来扩展它的计算能力
3. 高容错性
   Mapreduce的设计初衷就是使程序能够部署在廉价的pc机器上，这就要求它具有很高的容错性。比如一个机器挂了，它可以把上面的计算任务转移到另一个节点上运行，不至于这个任务运行失败，而且这个过程不需要人工参与，而完全是由hadoop内部完成的。
4. 适合PB级以上海量数据的离线处理

2. 缺点
文章图片

MapReduce虽然有很多的优势，但是也有它不擅长的。这里的“不擅长”，不代表不能做，而是在有些场景下实现的效果差，并不适合用MapReduce来处理，主要表现在以下结果方面：

1. 实时计算： MapReduce主要处理的数据来自于文件系统，所以无法像Oracle或MySQL那样在毫米或秒级内返回结果，如果需要大数据量的毫秒级响应，可以考虑结合实时存储系统来实现，利用HBase、Kudu等.
2. 流计算： 流计算的输入数据是动态的，而MapReduce主要的输入来自于HDFS等文件系统，数据是静态的，不能动态变化，这是因为MapReduce自身的设计特点决定了数据源必须是静态的。如果需要处理流式数据可以用Storm,Spark Steaming、Flink等流计算框架。
3. DGA（有向无环图）计算： 多个应用程序存在依赖关系，后一个应用程序的输入为前一个的输出。在这种情况下，MapReduce并不是不能做，而是使用后，每个MapReduce作业的输出结果都会写入磁盘，会造成大量的词频IO导致性能非常低下，此时可以考虑用Spark等迭代计算框架。

综合以上问题，MapReduce在解决离线分布式计算的过程中，主要考虑如何提升性能，加快分布式计算过程。
3. 需求 基于整个MapReduce所存在的缺点，由于MapReduce整体结构已经固定，所以整体的优化方案只能从以下两点来考虑实现：

1. 利用列存储思想，优化存储结构列存储在数据仓库、OLAP (on-line analytical processing)联机分析处理等应用上可以提高查询性能。利用列存储思想对MapReduce框架进行优化，面临合理的数据结构设计及数据压缩等挑战。

文章图片

2. 利用硬件资源优化连接算法，提高每个阶段的连接效率，MapReduce框架处理连接操作的过程比较复杂，面临数据倾斜、分布式环境数据传输及需要多个MapReduce作业等挑战，优化MapReduce每个阶段中的资源可以充分的利用硬件资源性能来提升MapReduce的效率。
   
   文章图片
   

二. IO性能优化：文件类型 1. 优化方案

1. 针对HDFS最初是为访问大文件而开发的, 所以会出现对大量小文件的存储效率不高问题, MapReduce在读取小文件进行处理时，也存在资源浪费导致计算效率不高的问题采用 SequenceFile和MapFile设计一个 HDFS中合并存储小文件的方案。该方案的主要思想是将小文件序列化存入一个 SequenceFIle/MapFile 容器,合并成大文件, 并建立相应的索引文件, 有效降低文件数目和提高访问效率. 通过和现有的 Hadoop Archives(HAR files)文件归档解决小文件问题的方案对比, 实验结果表明, 基于SequenceFile或者MapFile的存储小文件方案可以更为有效的提高小文件存储性能和减少HDFS文件系统的节点内存消耗
2. 针对普通按行存储文本文件，MapReduce在处理实现聚合、过滤等功能时，性能相对较差，针对行式存储的数据处理性能差的问题，可以选择使用列式存储的方案来实现数据聚合处理，降低数据传输及读写的IO，提高整体MapReduce计算处理的性能

2. SequenceFile
1. 介绍 SequenceFile是hadoop里用来存储序列化的键值对即二进制的一种文件格式。SequenceFile文件也可以作为MapReduce作业的输入和输出，hive和spark也支持这种格式。
它有如下几个优点：

• 以二进制的KV形式存储数据，与底层交互更加友好，性能更快，所以可以在HDFS里存储图像或者更加复杂的结构作为KV对。
• SequenceFile支持压缩和分片。 当你压缩为一个SequenceFile时，并不是将整个文件压缩成一个单独的单元，而是压缩文件里的record或者block of records（块）。因此SequenceFile是能够支持分片的，即使使用的压缩方式如Snappy, Lz4 or Gzip不支持分片，也可以利用SequenceFIle来实现分片。
• SequenceFile也可以用于存储多个小文件。由于Hadoop本身就是用来处理大型文件的，小文件是不适合的，所以用一个SequenceFile来存储很多小文件就可以提高处理效率，也能节省Namenode内存，因为Namenode只需一个SequenceFile的metadata，而不是为每个小文件创建单独的metadata。
• 由于数据是以SequenceFile形式存储，所以中间输出文件即map输出也会用SequenceFile来存储，可以提高整体的IO开销性能

2. 存储特点
文章图片

1. sequenceFile文件是Hadoop用来存储二进制形式的[Key,Value]对而设计的一种平面文件(Flat File)。
2. 可以把SequenceFile当做是一个容器，把所有的文件打包到SequenceFile类中可以高效的对小文件进行存储和处理。
3. SequenceFile文件并不按照其存储的Key进行排序存储，SequenceFile的内部类Writer提供了append功能。
4. SequenceFile中的Key和Value可以是任意类型Writable或者是自定义Writable。
5. 存储结构上，SequenceFile主要由一个Header后跟多条Record组成，Header主要包含了Key classname，value classname，存储压缩算法，用户自定义元数据等信息，此外，还包含了一些同步标识，用于快速定位到记录的边界。每条Record以键值对的方式进行存储，用来表示它的字符数组可以一次解析成：记录的长度、Key的长度、Key值和value值，并且Value值的结构取决于该记录是否被压缩。
6. 在recourds中，又分为是否压缩格式。当没有被压缩时，key与value使用Serialization序列化写入SequenceFile。当选择压缩格式时，record的压缩格式与没有压缩其实不尽相同，除了value的bytes被压缩，key是不被压缩的
7. 在Block中，它使所有的信息进行压缩，压缩的最小大小由配置文件中io.seqfile.compress.blocksize配置项决定。

3. SequenceFile 工具类

1. SequenceFileOutputFormat
   用于将MapReduce的结果输出为SequenceFile文件
2. SequenceFileInputFormat
   用于读取SequenceFile文件

4. 生成SequenceFile

1. 需求:将普通文件转换为SequenceFile文件
   
   文章图片
   
2. 思路
   
   • 使用TextInputFormat读取普通文件文件
   • Map阶段对读取文件的每一行进行输出
   • Reduce阶段直接输出每条数据
   • 使用SequenceFileOutputFormat将结果保存为SequenceFile
3. 代码实现
   
   • Driver类

   import org.apache.hadoop.conf.Configuration; import org.apache.hadoop.conf.Configured; import org.apache.hadoop.fs.Path; import org.apache.hadoop.io.*; import org.apache.hadoop.mapreduce.Job; import org.apache.hadoop.mapreduce.Mapper; import org.apache.hadoop.mapreduce.Reducer; import org.apache.hadoop.mapreduce.lib.input.FileInputFormat; import org.apache.hadoop.mapreduce.lib.input.TextInputFormat; import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat; import org.apache.hadoop.mapreduce.lib.output.SequenceFileOutputFormat; import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat; import org.apache.hadoop.util.Tool; import org.apache.hadoop.util.ToolRunner; import java.io.IOException; import java.util.Iterator; /** * @ClassName MrWriteToSequenceFile * @Description TODO 读取文本文件，转换为SequenceFile文件 * @Create Byitcast */ public class MrWriteToSequenceFile extends Configured implements Tool { //构建、配置、提交一个 MapReduce的Job public int run(String[] args) throws Exception { // 实例化作业 Job job = Job.getInstance(this.getConf(), "MrWriteToSequenceFile"); // 设置作业的主程序 job.setJarByClass(this.getClass()); // 设置作业的输入为TextInputFormat（普通文本） job.setInputFormatClass(TextInputFormat.class); // 设置作业的输入路径 FileInputFormat.addInputPath(job, new Path(args[0])); // 设置Map端的实现类 job.setMapperClass(WriteSeqFileAppMapper.class); // 设置Map端输入的Key类型 job.setMapOutputKeyClass(NullWritable.class); // 设置Map端输入的Value类型 job.setMapOutputValueClass(Text.class); // 设置作业的输出为SequenceFileOutputFormat job.setOutputFormatClass(SequenceFileOutputFormat.class); // 使用SequenceFile的块级别压缩 SequenceFileOutputFormat.setOutputCompressionType(job, SequenceFile.CompressionType.BLOCK); // 设置Reduce端的实现类 job.setReducerClass(WriteSeqFileAppReducer.class); // 设置Reduce端输出的Key类型 job.setOutputKeyClass(NullWritable.class); // 设置Reduce端输出的Value类型 job.setOutputValueClass(Text.class); // 从参数中获取输出路径 Path outputDir = new Path(args[1]); // 如果输出路径已存在则删除 outputDir.getFileSystem(this.getConf() ).delete(outputDir, true); // 设置作业的输出路径 FileOutputFormat.setOutputPath(job, outputDir); // 提交作业并等待执行完成 return job.waitForCompletion(true) ? 0 : 1; } //程序入口，调用run public static void main(String[] args) throws Exception { //用于管理当前程序的所有配置 Configuration conf = new Configuration(); int status = ToolRunner.run(conf, new MrWriteToSequenceFile(), args); System.exit(status); } }

   
   
   • Mapper类

   /** * 定义Mapper类 */ public static class WriteSeqFileAppMapper extends Mapper{ private NullWritable outputKey; @Override protected void setup(Context context) throws IOException, InterruptedException { this.outputKey = NullWritable.get(); } @Override protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException { context.write(outputKey, value); } @Override protected void cleanup(Context context) throws IOException, InterruptedException { this.outputKey = null; }}

   
   
   • Reduce类

   /** * 定义Reduce类 */ public static class WriteSeqFileAppReducer extends Reducer{ private NullWritable outputKey; @Override protected void setup(Context context) throws IOException, InterruptedException { this.outputKey = NullWritable.get(); } @Override protected void reduce(NullWritable key, Iterable value, Context context) throws IOException, InterruptedException { Iterator iterator = value.iterator(); while (iterator.hasNext()) { context.write(outputKey, iterator.next()); } } @Override protected void cleanup(Context context) throws IOException, InterruptedException { this.outputKey = null; } }

   
   5. 读取SequenceFile
   • 需求: 将上一步转换好的SequenceFile再解析转换为普通文本文件内容
   • 【Hadoop 入门笔记 二十三 : MapReduce性能优化-文件类型】思路:
     
     1. 使用SequenceFileInputformat读取SequenceFile
     2. Map阶段直接输出每一条数据
     3. Reduce阶段直接输出每一条数据
     4. 使用TextOutputFormat将结果保存为普通文本文件
   • 代码实现
     
     • Driver 类

   import org.apache.hadoop.conf.Configuration; import org.apache.hadoop.conf.Configured; import org.apache.hadoop.fs.Path; import org.apache.hadoop.io.NullWritable; import org.apache.hadoop.io.Text; import org.apache.hadoop.mapreduce.Job; import org.apache.hadoop.mapreduce.Mapper; import org.apache.hadoop.mapreduce.Reducer; import org.apache.hadoop.mapreduce.lib.input.SequenceFileInputFormat; import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat; import org.apache.hadoop.util.Tool; import org.apache.hadoop.util.ToolRunner; import java.io.IOException; import java.util.Iterator; /** * @ClassName MrReadFromSequenceFile * @Description TODO 读取SequenceFile文件，转换为普通文本文件 * @Create Byitcast */ public class MrReadFromSequenceFile extends Configured implements Tool { //构建、配置、提交一个 MapReduce的Job public int run(String[] args) throws Exception { // 实例化作业 Job job = Job.getInstance(this.getConf(), "MrReadFromSequenceFile"); // 设置作业的主程序 job.setJarByClass(this.getClass()); // 设置作业的输入为SequenceFileInputFormat（SequenceFile文本） job.setInputFormatClass(SequenceFileInputFormat.class); // 设置作业的输入路径 SequenceFileInputFormat.addInputPath(job, new Path(args[0])); // 设置Map端的实现类 job.setMapperClass(ReadSeqFileAppMapper.class); // 设置Map端输入的Key类型 job.setMapOutputKeyClass(NullWritable.class); // 设置Map端输入的Value类型 job.setMapOutputValueClass(Text.class); // 设置作业的输出为TextOutputFormat job.setOutputFormatClass(TextOutputFormat.class); // 设置Reduce端的实现类 job.setReducerClass(ReadSeqFileAppReducer.class); // 设置Reduce端输出的Key类型 job.setOutputKeyClass(NullWritable.class); // 设置Reduce端输出的Value类型 job.setOutputValueClass(Text.class); // 从参数中获取输出路径 Path outputDir = new Path(args[1]); // 如果输出路径已存在则删除 outputDir.getFileSystem(this.getConf()).delete(outputDir, true); // 设置作业的输出路径 TextOutputFormat.setOutputPath(job, outputDir); // 提交作业并等待执行完成 return job.waitForCompletion(true) ? 0 : 1; } //程序入口，调用run public static void main(String[] args) throws Exception { //用于管理当前程序的所有配置 Configuration conf = new Configuration(); int status = ToolRunner.run(conf, new MrReadFromSequenceFile(), args); System.exit(status); } }

   
   
   • Mapper类

   /** * 定义Mapper类 */ public static class ReadSeqFileAppMapper extends Mapper { private NullWritable outputKey; @Override protected void setup(Context context) throws IOException, InterruptedException { this.outputKey = NullWritable.get(); } @Override protected void map(NullWritable key, Text value, Context context) throws IOException, InterruptedException { context.write(outputKey, value); } @Override protected void cleanup(Context context) throws IOException, InterruptedException { this.outputKey = null; }

   
   
   • Reducer类

   /** * 定义Reduce类 */ public static class ReadSeqFileAppReducer extends Reducer{ private NullWritable outputKey; @Override protected void setup(Context context) throws IOException, InterruptedException { this.outputKey = NullWritable.get(); } @Override protected void reduce(NullWritable key, Iterable value, Context context) throws IOException, InterruptedException { Iterator iterator = value.iterator(); while (iterator.hasNext()) { context.write(outputKey, iterator.next()); } } @Override protected void cleanup(Context context) throws IOException, InterruptedException { this.outputKey = null; } }

   
   

3. MapFile
1. 介绍 可以理解MapFile是排序后的SequenceFile， 通过观察其结构可以看到MapFile由两部分组成分别是data和index。data既存储数据的文件，index作为文件的数据索引，主要记录了每个Record的Key值，以及该Record在文件中的偏移位置。在MapFile被访问的时候，索引文件会被加载到内存，通过索引映射关系可以迅速定位到指定Record所在文件位置，因此，相对SequenceFile而言，MapFile的检索效率是最高的，缺点是会消耗一部分内存来存储index数据。
MapFile的数据存储结构：

文章图片


文章图片

2. MapFile工具类

1. MapFileOutputFormat：用于将MapReduce的结果输出为MapFile
2. MapReduce中没有封装MapFile的读取输入类，可根据情况选择以下方案来实现
   
   • 自定义InputFormat，使用MapFileOutputFormat中的getReader方法获取读取对象
   • 使用SequenceFileInputFormat对MapFile的数据进行解析
3. 生成MapFile文件
   
   • 需求: 将普通文件转换为MapFile文件
   • 思路:
     
     • Input读取一个普通文件
     • Map阶段构建随机值作为Key，构建有序
     • Output生成MapFile文件
   • 实现

   import org.apache.hadoop.conf.Configuration; import org.apache.hadoop.conf.Configured; import org.apache.hadoop.fs.Path; import org.apache.hadoop.io.*; import org.apache.hadoop.mapreduce.Job; import org.apache.hadoop.mapreduce.Mapper; import org.apache.hadoop.mapreduce.Reducer; import org.apache.hadoop.mapreduce.lib.input.FileInputFormat; import org.apache.hadoop.mapreduce.lib.input.TextInputFormat; import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat; import org.apache.hadoop.mapreduce.lib.output.MapFileOutputFormat; import org.apache.hadoop.mapreduce.lib.output.SequenceFileOutputFormat; import org.apache.hadoop.util.Tool; import org.apache.hadoop.util.ToolRunner; import java.io.IOException; import java.util.Iterator; import java.util.Random; /** * @ClassName MrWriteToMapFile * @Description TODO 读取文本文件，转换为MapFile文件 * @Create Byitcast */ public class MrWriteToMapFile extends Configured implements Tool { //构建、配置、提交一个 MapReduce的Job public int run(String[] args) throws Exception { Configuration conf = getConf(); // 实例化作业 Job job = Job.getInstance(conf, "MrWriteToMapFile"); // 设置作业的主程序 job.setJarByClass(this.getClass()); // 设置作业的输入为TextInputFormat（普通文本） job.setInputFormatClass(TextInputFormat.class); // 设置作业的输入路径 FileInputFormat.addInputPath(job, new Path(args[0])); // 设置Map端的实现类 job.setMapperClass(WriteMapFileAppMapper.class); // 设置Map端输入的Key类型 job.setMapOutputKeyClass(IntWritable.class); // 设置Map端输入的Value类型 job.setMapOutputValueClass(Text.class); // 设置作业的输出为MapFileOutputFormat job.setOutputFormatClass(MapFileOutputFormat.class); // 设置Reduce端的实现类 job.setReducerClass(WriteMapFileAppReducer.class); // 设置Reduce端输出的Key类型 job.setOutputKeyClass(IntWritable.class); // 设置Reduce端输出的Value类型 job.setOutputValueClass(Text.class); // 从参数中获取输出路径 Path outputDir = new Path(args[1]); // 如果输出路径已存在则删除 outputDir.getFileSystem(conf).delete(outputDir, true); // 设置作业的输出路径 MapFileOutputFormat.setOutputPath(job, outputDir); // 提交作业并等待执行完成 return job.waitForCompletion(true) ? 0 : 1; } //程序入口，调用run public static void main(String[] args) throws Exception { //用于管理当前程序的所有配置 Configuration conf = new Configuration(); int status = ToolRunner.run(conf, new MrWriteToMapFile(), args); System.exit(status); } /** * 定义Mapper类 */ public static class WriteMapFileAppMapper extends Mapper{ //定义输出的Key,每次随机生成一个值 private IntWritable outputKey = new IntWritable(); @Override protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException { //随机生成一个数值 Random random = new Random(); this.outputKey.set(random.nextInt(100000)); context.write(outputKey, value); } /** * 定义Reduce类 */ public static class WriteMapFileAppReducer extends Reducer{@Override protected void reduce(IntWritable key, Iterable value, Context context) throws IOException, InterruptedException { Iterator iterator = value.iterator(); while (iterator.hasNext()) { context.write(key, iterator.next()); } } }}

   
   
   4. 读取MapFile文件
      
      • 需求：将MapFile解析为普通文件内容
      • 思路:
        
        • input读取MapFile，注意，Hadoop没有提供MapFileInputFormat，所以使用SequenceFileInputFormat来解析，或者可以自定义InputFormat
        • Map和Reduce直接输出
        • Output将结果保存为普通文件
      • 实现

   import org.apache.hadoop.conf.Configuration; import org.apache.hadoop.conf.Configured; import org.apache.hadoop.fs.Path; import org.apache.hadoop.io.IntWritable; import org.apache.hadoop.io.NullWritable; import org.apache.hadoop.io.Text; import org.apache.hadoop.mapreduce.Job; import org.apache.hadoop.mapreduce.Mapper; import org.apache.hadoop.mapreduce.Reducer; import org.apache.hadoop.mapreduce.lib.input.FileInputFormat; import org.apache.hadoop.mapreduce.lib.input.SequenceFileInputFormat; import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat; import org.apache.hadoop.util.Tool; import org.apache.hadoop.util.ToolRunner; import java.io.IOException; import java.util.Iterator; /** * @ClassName MrReadFromMapFile * @Description TODO 读取MapFile文件，转换为普通文本文件 * @Create Byitcast */ public class MrReadFromMapFile extends Configured implements Tool { //构建、配置、提交一个 MapReduce的Job public int run(String[] args) throws Exception { // 实例化作业 Job job = Job.getInstance(this.getConf(), "MrReadFromMapFile"); // 设置作业的主程序 job.setJarByClass(this.getClass()); // 设置作业的输入为SequenceFileInputFormat（Hadoop没有直接提供MapFileInput） job.setInputFormatClass(SequenceFileInputFormat.class); // 设置作业的输入路径 SequenceFileInputFormat.addInputPath(job, new Path(args[0])); // 设置Map端的实现类 job.setMapperClass(ReadMapFileAppMapper.class); // 设置Map端输入的Key类型 job.setMapOutputKeyClass(NullWritable.class); // 设置Map端输入的Value类型 job.setMapOutputValueClass(Text.class); // 设置作业的输出为SequenceFileOutputFormat job.setOutputFormatClass(TextOutputFormat.class); // 设置Reduce端的实现类 job.setReducerClass(ReadMapFileAppReducer.class); // 设置Reduce端输出的Key类型 job.setOutputKeyClass(NullWritable.class); // 设置Reduce端输出的Value类型 job.setOutputValueClass(Text.class); // 从参数中获取输出路径 Path outputDir = new Path(args[1]); // 如果输出路径已存在则删除 outputDir.getFileSystem(this.getConf()).delete(outputDir, true); // 设置作业的输出路径 TextOutputFormat.setOutputPath(job, outputDir); // 提交作业并等待执行完成 return job.waitForCompletion(true) ? 0 : 1; } //程序入口，调用run public static void main(String[] args) throws Exception { //用于管理当前程序的所有配置 Configuration conf = new Configuration(); int status = ToolRunner.run(conf, new MrReadFromMapFile(), args); System.exit(status); } /** * 定义Mapper类 */ public static class ReadMapFileAppMapper extends Mapper {private NullWritable outputKey = NullWritable.get(); @Override protected void map(IntWritable key, Text value, Context context) throws IOException, InterruptedException { context.write(outputKey, value); } } /** * 定义Reduce类 */ public static class ReadMapFileAppReducer extends Reducer{@Override protected void reduce(NullWritable key, Iterable value, Context context) throws IOException, InterruptedException { Iterator iterator = value.iterator(); while (iterator.hasNext()) { context.write(key, iterator.next()); } } }}

   
   

4. ORCFile
1. ORC介绍 ORC（OptimizedRC File）文件格式是一种Hadoop生态圈中的列式存储格式，源自于RC（RecordColumnar File），它的产生早在2013年初，最初产生自Apache Hive，用于降低Hadoop数据存储空间和加速Hive查询速度。它并不是一个单纯的列式存储格式，仍然是首先根据行组分割
整个表，在每一个行组内进行按列存储。ORC文件是自描述的，它的元数据使用Protocol Buffers序列化，并且文件中的数据尽可能的压缩以降低存储空间的消耗，目前也被Spark SQL、Presto等查询引擎支持。2015年ORC项目被Apache项目基金会提升为Apache顶级项目。

文章图片

ORC文件中保存了三个层级的统计信息，分别为文件级别、stripe级别和row group级别的，他们都可以用来根据Search ARGuments（谓词下推条件）判断是否可以跳过某些数据，在统计信息中都包含成员数和是否有null值，并且对于不同类型的数据设置一些特定的统计信息。
性能测试:

• 原始Text格式，未压缩 : 38.1 G
• ORC格式，默认压缩（ZLIB）: 11.5 G
• Parquet格式，默认压缩（Snappy）：14.8 G
• 测试对比：复杂数据Join关联测试
  
  文章图片
  

2. ORCFile工具类 1. 添加ORC与MapReduce集成的Maven依赖

org.apache.orc orc-mapreduce 1.6.3

2. 生成ORC文件

1. 需求: 将普通文件转换为ORC文件
   
   文章图片
   
2. 实现思路:
   
   • Input阶段读取普通文件
   • Map:阶段直接输出数据，没有Reduce阶段
   • Output阶段使用OrcOutputFormat保存为ORC文件类型
3. 实现
   

   import org.apache.hadoop.conf.Configuration; import org.apache.hadoop.conf.Configured; import org.apache.hadoop.fs.Path; import org.apache.hadoop.io.LongWritable; import org.apache.hadoop.io.NullWritable; import org.apache.hadoop.io.Text; 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.TextInputFormat; import org.apache.hadoop.util.Tool; import org.apache.hadoop.util.ToolRunner; import org.apache.orc.OrcConf; import org.apache.orc.TypeDescription; import org.apache.orc.mapred.OrcStruct; import org.apache.orc.mapreduce.OrcOutputFormat; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import java.io.IOException; /** * @ClassName WriteOrcFileApp * @Description TODO 用于读取普通文本文件转换为ORC文件 */ public class WriteOrcFileApp extends Configured implements Tool { // 作业名称 private static final String JOB_NAME = WriteOrcFileApp.class.getSimpleName(); //构建日志监听 private static final Logger LOG = LoggerFactory.getLogger(WriteOrcFileApp.class); //定义数据的字段信息 private static final String SCHEMA = "struct"; /** * 重写Tool接口的run方法，用于提交作业 * @param args * @return * @throws Exception */ public int run(String[] args) throws Exception { // 设置Schema OrcConf.MAPRED_OUTPUT_SCHEMA.setString(this.getConf(), SCHEMA); // 实例化作业 Job job = Job.getInstance(this.getConf(), JOB_NAME); // 设置作业的主程序 job.setJarByClass(WriteOrcFileApp.class); // 设置作业的Mapper类 job.setMapperClass(WriteOrcFileAppMapper.class); // 设置作业的输入为TextInputFormat（普通文本） job.setInputFormatClass(TextInputFormat.class); // 设置作业的输出为OrcOutputFormat job.setOutputFormatClass(OrcOutputFormat.class); // 设置作业使用0个Reduce（直接从map端输出） job.setNumReduceTasks(0); // 设置作业的输入路径 FileInputFormat.addInputPath(job, new Path(args[0])); // 从参数中获取输出路径 Path outputDir = new Path(args[1]); // 如果输出路径已存在则删除 outputDir.getFileSystem(this.getConf()).delete(outputDir, true); // 设置作业的输出路径 OrcOutputFormat.setOutputPath(job, outputDir); // 提交作业并等待执行完成 return job.waitForCompletion(true) ? 0 : 1; } //程序入口，调用run public static void main(String[] args) throws Exception { //用于管理当前程序的所有配置 Configuration conf = new Configuration(); int status = ToolRunner.run(conf, new WriteOrcFileApp(), args); System.exit(status); } /** * 实现Mapper类 */ public static class WriteOrcFileAppMapper extends Mapper { //获取字段描述信息 private TypeDescription schema = TypeDescription.fromString(SCHEMA); //构建输出的Key private final NullWritable outputKey = NullWritable.get(); //构建输出的Value为ORCStruct类型 private final OrcStruct outputValue = https://www.it610.com/article/(OrcStruct) OrcStruct.createValue(schema); public void map(LongWritable key, Text value, Context output) throws IOException, InterruptedException { //将读取到的每一行数据进行分割，得到所有字段 String[] fields = value.toString().split(",",8); //将所有字段赋值给Value中的列 outputValue.setFieldValue(0, new Text(fields[0])); outputValue.setFieldValue(1, new Text(fields[1])); outputValue.setFieldValue(2, new Text(fields[2])); outputValue.setFieldValue(3, new Text(fields[3])); outputValue.setFieldValue(4, new Text(fields[4])); outputValue.setFieldValue(5, new Text(fields[5])); outputValue.setFieldValue(6, new Text(fields[6])); outputValue.setFieldValue(7, new Text(fields[7])); //输出KeyValue output.write(outputKey, outputValue); } } }

   
   

问题:
报错：缺少orc-mapreduce的jar包依赖

文章图片

解决：将orc-MapReduce的jar包添加到Hadoop的环境变量中，所有NodeManager节点都要添加

cporc-shims-1.6.3.jarorc-core-1.6.3.jarorc-mapreduce-1.6.3.jar aircompressor-0.15.jarhive-storage-api-2.7.1.jar /export/server/hadoop-3.1.4/share/hadoop/mapreduce/

文章图片

3. 读取ORC文件

1. 需求: 读取ORC文件，还原成普通文本文件
2. 思路:
   
   • Input阶段读取上一步当中生成的ORC文件
   • Map阶段直接读取输出
   • Output阶段将结果保存为普通文本文件
3. 实现
   

   import org.apache.hadoop.conf.Configuration; import org.apache.hadoop.conf.Configured; import org.apache.hadoop.fs.Path; import org.apache.hadoop.io.NullWritable; import org.apache.hadoop.io.Text; import org.apache.hadoop.mapreduce.Job; import org.apache.hadoop.mapreduce.Mapper; import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat; import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat; import org.apache.hadoop.util.Tool; import org.apache.hadoop.util.ToolRunner; import org.apache.orc.mapred.OrcStruct; import org.apache.orc.mapreduce.OrcInputFormat; import java.io.IOException; /** * @ClassName ReadOrcFileApp * @Description TODO 读取ORC文件进行解析还原成普通文本文件 */ public class ReadOrcFileApp extends Configured implements Tool { // 作业名称 private static final String JOB_NAME = WriteOrcFileApp.class.getSimpleName(); /** * 重写Tool接口的run方法，用于提交作业 * @param args * @return * @throws Exception */ public int run(String[] args) throws Exception { // 实例化作业 Job job = Job.getInstance(this.getConf(), JOB_NAME); // 设置作业的主程序 job.setJarByClass(ReadOrcFileApp.class); // 设置作业的输入为OrcInputFormat job.setInputFormatClass(OrcInputFormat.class); // 设置作业的输入路径 OrcInputFormat.addInputPath(job, new Path(args[0])); // 设置作业的Mapper类 job.setMapperClass(ReadOrcFileAppMapper.class); // 设置作业使用0个Reduce（直接从map端输出） job.setNumReduceTasks(0); // 设置作业的输入为TextOutputFormat job.setOutputFormatClass(TextOutputFormat.class); // 从参数中获取输出路径 Path outputDir = new Path(args[1]); // 如果输出路径已存在则删除 outputDir.getFileSystem(this.getConf()).delete(outputDir, true); // 设置作业的输出路径 FileOutputFormat.setOutputPath(job, outputDir); // 提交作业并等待执行完成 return job.waitForCompletion(true) ? 0 : 1; } //程序入口，调用run public static void main(String[] args) throws Exception { //用于管理当前程序的所有配置 Configuration conf = new Configuration(); int status = ToolRunner.run(conf, new ReadOrcFileApp(), args); System.exit(status); } /** * 实现Mapper类 */ public static class ReadOrcFileAppMapper extends Mapper { private NullWritable outputKey; private Text outputValue; @Override protected void setup(Context context) throws IOException, InterruptedException { outputKey = NullWritable.get(); outputValue = https://www.it610.com/article/new Text(); } public void map(NullWritable key, OrcStruct value, Context output) throws IOException, InterruptedException { //将ORC中的每条数据转换为Text对象 this.outputValue.set( value.getFieldValue(0).toString()+","+ value.getFieldValue(1).toString()+","+ value.getFieldValue(2).toString()+","+ value.getFieldValue(3).toString()+","+ value.getFieldValue(4).toString()+","+ value.getFieldValue(5).toString()+","+ value.getFieldValue(6).toString()+","+ value.getFieldValue(7).toString() ); //输出结果 output.write(outputKey, outputValue); } } }

   
   

推荐阅读

• 

  可以推荐几部非常好看的电视剧吗？
• 

  寸草心谜底揭秘
• 

  去4s店检测故障要钱吗多少钱 去4s店检测故障要钱吗
• 

  mysql版本号怎么查看 mysql 查版本
• 

  鱼汤馄饨做法步骤视频 鱼汤馄饨做法步骤
• 

  遇到西门子洗衣机显示bb怎么办,主要差别在这里
• 

  人工智能 人员分析
• 

  生吃鱼腥草有寄生虫吗
• 

  阴沉木鉴定方法 阴沉木鉴定方法
• 

  在你心里谁是真正的演技派和实力派？
• 

  红米手机怎么关闭流量套餐
• 

  决眦入归鸟上一句是什么
• 

  恐龙的体型为什么那么大
• 

  精通分析师教材
• 

  永远的拼音 永的拼音
• 

  OPPOA73如何使用手势导航 具体操作流程介绍
• 

  万字的笔顺 汉字万字的笔顺
• 

  什么叫代拍摄影，摄影代拍平台
• 

  红豆薏米绿豆的功效
• 

  原理图分析答辩,偿债能力分析答辩问题

• EffectiveObjective-C2.0|EffectiveObjective-C2.0 笔记 - 第二部分
• Android中的AES加密-下
• 【读书笔记】贝叶斯原理
• 【韩语学习】（韩语随堂笔记整理）
• 【Hadoop踩雷】Mac下安装Hadoop3以及Java版本问题
• 人性的弱点-笔记
• 读书笔记:博登海默法理学|读书笔记:博登海默法理学 —— 正义的探索（1）
• D034+3组苏曼＋《写作这回事》读书笔记
• 《自我的追寻》读书笔记3
• 最有效的时间管理工具（赢效率手册和总结笔记）