其他分享
首页 > 其他分享> > Hadoop 性能优化

Hadoop 性能优化

作者:互联网

Hadoop 性能优化

小文件问题

HDFS和MapReduce是针对大文件设计的,在小文件处理上效率低下,且十分消耗内存资源。每个小文件都会占用一个block、产生一个InputSplit、产生一个Map任务,这样map任务的启动时间很长,执行任务的时间很短。解决方法是使用容器将小文件组织起来,HDFS提供了两种容器:SequenceFile MapFile

  1. SequenceFile
    SequeceFile是Hadoop提供的一种二进制文件,这种二进制 文件直接将<key, value>对序列化到文件中
    一般对小文件可以使用这种文件合并,即将文件名作为key, 文件内容作为value序列化到大文件中
    注意:SequeceFile需要一个合并文件的过程,文件较大,且合并后的文件将不方便查看,必须通过遍历查看每一个小文件
    SequenceFile
    package org.example.mapreduce;
    
    import org.apache.commons.io.FileUtils;
    import org.apache.hadoop.conf.Configuration;
    import org.apache.hadoop.fs.FileSystem;
    import org.apache.hadoop.fs.Path;
    import org.apache.hadoop.io.SequenceFile;
    import org.apache.hadoop.io.Text;
    
    import java.io.File;
    import java.nio.charset.StandardCharsets;
    
    public class SmallFileSeq {
    
        public static void main(String[] args) throws Exception {
            write("D:\\samllFile", "/SequenceFile");
        }
    
        /**
         * 生成 SequenceFile 文件
         * @param inputDir 本地存放小文件的目录
         * @param outPutFile 输出压缩文件的 hdfs 目录
         * @throws Exception
         */
        private static void write(String inputDir, String outPutFile) throws Exception {
            Configuration conf = new Configuration();
            conf.set("fs.defaultFS", "hdfs://ip:9000");
    
    
            // 如果输出文件存在则删除
            FileSystem fileSystem = FileSystem.get(conf);
            fileSystem.delete(new Path(outPutFile), true);
    
    
            // 构造 option 数组,有三个元素
            // 1. 输出路径
            // 2. key 类型
            // 3. value 类型
            SequenceFile.Writer.Option[] option = new SequenceFile.Writer.Option[]{
                    SequenceFile.Writer.file(new Path(outPutFile)),
                    SequenceFile.Writer.keyClass(Text.class),
                    SequenceFile.Writer.valueClass(Text.class)
            };
    
            // 创建一个 Write 实例
            SequenceFile.Writer writer = SequenceFile.createWriter(conf, option);
    
            // 指定要压缩的文件的目录
            File inputDirPath = new File(inputDir);
            if (inputDirPath.isDirectory()) {
                for (File file : inputDirPath.listFiles()) {
                    // 获取文件全部,对于小文件直接全部读取到内存
                    String content = FileUtils.readFileToString(file, StandardCharsets.UTF_8);
                    // 获取文件名
                    String fileName = file.getName();
                    writer.append(new Text(fileName), new Text(content));
                }
            }
    
            writer.close();
        }
    
        /**
         * 读取 SequenceFile
         * @param inputFile
         * @throws Exception
         */
        private static void read(String inputFile) throws Exception {
            Configuration conf = new Configuration();
            conf.set("fs.defaultFS", "hdfs://ip:9000");
    
            // 创建阅读器
            SequenceFile.Reader reader = new SequenceFile.Reader(conf, SequenceFile.Reader.file(new Path(inputFile)));
    
            // 循环读取数据
            Text key = new Text();
            Text value = new Text();
            while (reader.next(key, value)) {
                System.out.println("文件名 : " + key);
                System.out.println("文件内容 : " + value);
            }
    
            reader.close();
        }
    }

    在 mapreduce 任务中需要指定输入处理类,重新指定 map 函数的key

    操作SequenceFile的MapReduce
    public class WordCountJobSeq {
        public static void main(String[] args) {
            try {
                if (args.length != 2) {
                    System.exit(1);
                }
    
                // 创建一个配置类
                Configuration conf = new Configuration();
                // 创建一个任务
                Job job = Job.getInstance(conf);
                
                
    
                // ... 省略
                
    
                
                // 默认情况下使用Text处理类,当处理 SequenceFile 时需要指定处理类
                job.setInputFormatClass(SequenceFileInputFormat.class);
    
                
    
                // 提交 job
                job.waitForCompletion(true);
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
    
        // 泛型的第一个参数和map函数的第一个参数默认应该为LongWriter用于表示每行字节偏移量
        // 读取SequenceFile时应为Text类型,值为压缩包内文件名
        public static class  MyMapper extends Mapper<Text, Text, Text, LongWritable> {
            @Override
            protected void map(Text key, // 默认情况下key为偏移字节数,读取 SequenceFile 时key为Text类型、内容为文件名
                               Text value,
                               Mapper<Text, Text, Text, LongWritable>.Context context
            ) throws IOException, InterruptedException {
                // ...
            }
        }
    
        // reduce 相同
        public static class MyReducer extends Reducer<Text, LongWritable, Text, LongWritable> {
            @Override
            protected void reduce(
                    Text key,
                    Iterable<LongWritable> values,
                    Reducer<Text, LongWritable, Text, LongWritable>.Context context
            ) throws IOException, InterruptedException {
                 // ...
            }
        }
    }
  2. MapFile
    MapFile是排序后的SequenceFile , MapFile由两部分组成, 分别是index和data
    index作为文件的数据索引,主要记录了每个Record的key值, 以及该Record在文件中的偏移位置
    在MapFile被访问的时候,索引文件会被加载到内存,通过索引 映射关系可迅速定位到指定Record所在文件位置
    优点:检索效率高
    缺点:要消耗一部分内存用于存储index索引
    MapFile
    package org.example.mapreduce;
    
    import org.apache.commons.io.FileUtils;
    import org.apache.hadoop.conf.Configuration;
    import org.apache.hadoop.fs.FileSystem;
    import org.apache.hadoop.fs.Path;
    import org.apache.hadoop.io.MapFile;
    import org.apache.hadoop.io.SequenceFile;
    import org.apache.hadoop.io.Text;
    
    import java.io.File;
    import java.nio.charset.StandardCharsets;
    
    public class SmallFileMap {
    
        public static void main(String[] args) throws Exception {
            write("D:\\samllFile", "/MapFile");
            read("/MapFile");
        }
    
        /**
         * 生成 MapFile 文件
         * @param inputDir 本地存放小文件的目录
         * @param outPutDir 输出压缩文件的 hdfs 目录,其下有两个文件 一个index索引 一个数据文件
         * @throws Exception
         */
        private static void write(String inputDir, String outPutDir) throws Exception {
            Configuration conf = new Configuration();
            conf.set("fs.defaultFS", "hdfs://ip:9000");
    
    
            // 如果输出文件存在则删除
            FileSystem fileSystem = FileSystem.get(conf);
            fileSystem.delete(new Path(outPutDir), true);
    
    
            // 构造 option 数组,有两个元素
            // 1. key 类型
            // 2. value 类型
            SequenceFile.Writer.Option[] options = new SequenceFile.Writer.Option[]{
                    MapFile.Writer.keyClass(Text.class),
                    MapFile.Writer.valueClass(Text.class)
            };
    
            // 创建一个 Write 实例
            MapFile.Writer writer = new MapFile.Writer(conf, new Path(outPutDir), options);
    
            // 指定要压缩的文件的目录
            File inputDirPath = new File(inputDir);
            if (inputDirPath.isDirectory()) {
                for (File file : inputDirPath.listFiles()) {
                    // 获取文件全部,对于小文件直接全部读取到内存
                    String content = FileUtils.readFileToString(file, StandardCharsets.UTF_8);
                    // 获取文件名
                    String fileName = file.getName();
                    writer.append(new Text(fileName), new Text(content));
                }
            }
    
            writer.close();
        }
    
        /**
         * 读取 MapFile
         * @param inputFile 读取 MapFile 文件路径
         * @throws Exception
         */
        private static void read(String inputFile) throws Exception {
            Configuration conf = new Configuration();
            conf.set("fs.defaultFS", "hdfs://ip:9000");
    
            // 创建阅读器
            MapFile.Reader reader = new MapFile.Reader(new Path(inputFile), conf);
    
            // 循环读取数据
            Text key = new Text();
            Text value = new Text();
            while (reader.next(key, value)) {
                System.out.println("文件名 : " + key);
                System.out.println("文件内容 : " + value);
            }
    
            reader.close();
        }
    }

 

数据倾斜问题

为了提高效率启动多个Reduce进程并行,这时涉及到将数据分区不到不同进程。可以通过  job.getPartitionerClass() 和 job.setPartitionerClass() 查询和设置分区类。默认使用HashPartitioner.class

job.getPartitionerClass();

// getPartitionerClass 源码
  public Class<? extends Partitioner<?,?>> getPartitionerClass() 
     throws ClassNotFoundException {
    return (Class<? extends Partitioner<?,?>>) 
      conf.getClass(PARTITIONER_CLASS_ATTR, HashPartitioner.class);
  }

// 分区类默认使用这个
public class HashPartitioner<K, V> extends Partitioner<K, V> {
  public int getPartition(K key, V value,
                          int numReduceTasks) {
    return (key.hashCode() & Integer.MAX_VALUE) % numReduceTasks;
  }

}

MapReduce程序执行时,Reduce节点大部分执行完毕,但是有一个或者几个Reduce节点运行很慢,导致整个程序处理 时间变得很长,具体表现为:Ruduce阶段一直卡着不动。

例子:一个 wordcount 任务,数据为0-9 的十个数字,其中数字"5"有910w个,其余9个数字有90w个。这时就叫数据发生了倾斜。为了加快速度,启动多个reduce任务。在reduce阶段,数字“5”会被分给同一个reduce任务执行,这个reduce任务的执行速度就会比其他的慢。

解决方法:

  1. 增加Reduce任务个数(针对数据倾斜不是太严重的情况有效)
  2. 把倾斜的数据打散
        public static class  MyMapper extends Mapper<LongWritable, Text, Text, LongWritable> {
            @Override
            protected void map(LongWritable key,
                               Text value,
                               Mapper<LongWritable, Text, Text, LongWritable>.Context context
            ) throws IOException, InterruptedException {
                // 分割单词
                String[] words = value.toString().split(" ");
                // 迭代分出的单词
                for (String word : words) {
                    // 把迭代出的单词封装为 <k2, v2> 的形式
                    
                    /*****************************************************************************/
                    
                    if ("5".equals(word)) { // 数据向字符“5”倾斜,这里将数据打散
                        word = "5" + "_" + new Random().nextInt(10); // 将 5 变为 5_0 ~ 5_9
                    }
                    
                    // 在得到的结果中key也会变化,之后需要再使用mapreduce将数据聚合
    
                    /*****************************************************************************/
                    
                    Text keyOut = new Text(word);
                    LongWritable valueOut = new LongWritable(1L);
                    // 把 <k2, v2> 写出去
                    context.write(keyOut, valueOut);
                }
            }
        }

    结果类似这样,需要再来一个程序进行处理。

标签:文件,conf,Text,性能,Hadoop,SequenceFile,key,new,优化
来源: https://www.cnblogs.com/zhh567/p/16339777.html