Сжатие

> Сжатие файлов имеет две основные выгоды: уменьшение занимаемого пространства для хранения файлов и ускорение передачи данных по сети и на диске.
 В Hadoop можно использовать множество форматов сжатия файлов, инструментов и алгоритмов, каждый из которых имеет свои преимущества.
 ! [Сводная таблица форматов сжатия](https://git.oschina.net/weiwei02/WHadoop/raw/master/doc/io/1.png) Таблица 1
 Все алгоритмы сжатия требуют балансировки между пространством и временем: быстрое сжатие и распаковка означает меньшее сжатие пространства. Все инструменты, перечисленные в Таблице 1, предоставляют 9 различных уровней для контроля времени и пространства при сжатии: опция -1 оптимизирует скорость сжатия, а опция -9 оптимизирует сжатие пространства. Например, мы можем создать сжатый файл с именем compress.gz с максимальным сжатием пространства следующей командой.
      gzip -9 compress
 Разные инструменты сжатия имеют разные характеристики. Gzip — это универсальный инструмент сжатия, который находится между двумя другими алгоритмами в балансировке пространства и времени. Сила сжатия bzip2 превосходит gzip, но скорость сжатия меньше. Хотя скорость распаковки bzip2 выше, чем скорость сжатия, она все равно медленнее, чем другие инструменты распаковки. LZO, LZ4 и Snappy — три метода сжатия, которые больше всего заботятся о скорости сжатия.Их скорость сжатия превосходит gzip, но коэффициент сжатия файлов ниже, чем у gzip. Кроме того, скорость распаковки Snappy и LZ4 значительно выше, чем у LZO.

##### 2. Определение CompressionCodec с помощью CompressionCodecFactory
В обычных сжатых файлах можно определить используемый алгоритм сжатия по расширению имени файла. Например, если расширение файла .gz, то для его распаковки можно использовать GzipCodec. CompressionCodecFactory предоставляет метод getCodec(), который позволяет отображать расширение имени файла на соответствующий объект CompressionCodec.

```java
/**
 * Находит соответствующий объект CompressionCodec для заданного файла на основе его
 * расширения имени файла.
 * @param file имя файла для проверки
 * @return объект codec
 */
public CompressionCodec getCodec(Path file) {
  CompressionCodec result = null;
  if (codecs != null) {
    String filename = file.getName();
    String reversedFilename = new StringBuilder(filename).reverse().toString();
    SortedMap<String, CompressionCodec> subMap = codecs.headMap(reversedFilename);
    if (!subMap.isEmpty()) {
      String potentialSuffix = subMap.lastKey();
      if (reversedFilename.startsWith(potentialSuffix)) {
        result = codecs.get(potentialSuffix);
      }
    }
  }
  return result;
}

пакет com.weiwei.WHadoop.io;
импорт org.apache.hadoop.fs.FileStatus;
импорт org.apache.hadoop.fs.Path;
импорт org.apache.hadoop.io.IOUtils;
импорт org.apache.hadoop.io.compress.CompressionCodec;
импорт org.apache.hadoop.io.compress.CompressionCodecFactory;
импорт org.apache.hadoop.io.compress.CompressionOutputStream;
```         import org.apache.hadoop.io.compress.GzipCodec;
          import org.apache.hadoop.util.ReflectionUtils;
          import org.junit.Test;
          import java.io.IOException;
          import java.io.InputStream;
          import java.io.OutputStream;
          /**
           * @author WangWeiwei
           * @version 1.0
           * @sine 17-2-28
           */
          public class FileDecompressorTest extends MiniClusterSuperTest {
              @Test
              public void testFileDecompressor() throws IOException {
                  Path path = new Path("/temp/a.gz");
                  CompressionCodec compressionCodec = ReflectionUtils.newInstance(GzipCodec.class, conf);
                  OutputStream outputStream = fs.create(path);
                  CompressionOutputStream compressionOutputStream = compressionCodec.createOutputStream(outputStream);
                  compressionOutputStream.write("abcdefghigh".getBytes());
                  compressionOutputStream.finish();
                  compressionOutputStream.close();
                  System.out.println("Сжатый размер файла: " + fs.getFileStatus(path).getLen());
                  CompressionCodecFactory factory = new CompressionCodecFactory(conf);
                  CompressionCodec codec = factory.getCodec(path);
                  if (codec == null) {
                      System.out.println("Не удалось найти метод сжатия файла");
                  } else {
                      InputStream in = null;
                      OutputStream out = null;
                      try {
                          Path target = new Path(factory.removeSuffix(path.toString(), codec.getDefaultExtension()));
                          in = codec.createInputStream(fs.open(path));
                          out = fs.create(target);
                          IOUtils.copyBytes(in, out, conf);
                          FileStatus fileStatus = fs.getFileStatus(target);
                          System.out.println("Размер распакованного файла: " + fileStatus.getLen());
                      } finally {
                          IOUtils.closeStream(in);
                          IOUtils.closeStream(out);
                      }
                  }
              }
          }
      ##### 3. CodecPool сжатия и распаковки    CodecPool может многократно использовать большое количество операций сжатия и распаковки, как это делается с пулеметами подключения к базе данных, что позволяет снизить затраты на создание объектов. Однако пулы сжатия/распаковки обычно используются в сценариях, где требуется большое количество операций сжатия и распаковки, а для небольшого количества операций сжатия/распаковки это совершенно не обязательно.
     Пример 3: Использование пула сжатия для сжатия данных, считанных из стандартного ввода, а затем запись их в стандартный вывод.
           package com. weiwei. WHadoop. io;
           import org. apache. hadoop. io. IOUtils;
           import org. apache. hadoop. io. compress. *;
           import org. junit. Test;
           import java. io. IOException;
           /**
           * @author WangWeiwei
           * @since OnClickListener 2017/3/1
           * Использование пула сжатия для сжатия данных, считанных из стандартного ввода, а затем запись их в стандартный вывод.
           */```markdown
         public class PooledStreamCompressorTest extends MiniClusterSuperTest {
               @Test
              public void testPooledStreamCompressor() {
                  CompressionCodec compressionCodec = new GzipCodec();
                  Compressor compressor = null;
                  try {
                      compressor = CodecPool. getCompressor(compressionCodec, conf);
                      CompressionOutputStream outputStream = compressionCodec. createOutputStream(System. out);
                      IOUtils. copyBytes(System. in, outputStream, 4096, false);
                      outputStream. finish();
                  } catch (IOException e) {
                      e. printStackTrace();
                  } finally {
                      CodecPool. returnCompressor(compressor);
                  }
              }
           }
 ```### 2. Сжатие и входные фрагменты   Для MapReduce особенно важно, поддерживает ли сжатие файлов возможность фрагментации. Если сжатый формат, например, Gzip, не поддерживает фрагментацию, так как gzip не поддерживает случайного доступа к данным, то для декомпрессии файлов требуется объединение нескольких файлов в один полный файл перед декомпрессией. Однако MapReduce требует свойства локальности данных, поэтому сжатие gzip в MapReduce требует объединения всех данных в один полный сжатый файл перед декомпрессией. Несмотря на высокую эффективность сжатия gzip, из вышеизложенного следует, что он не подходит для MapReduce.
   Существует два способа сжатия выходных данных MapReduce:
   1. Можно установить свойство mapred.output.compress в конфигурации MapReduce на true, а затем задать свойство mapred.output.compression.codec на полный путь к типу используемого сжатого codec.
   2. Можно использовать два статических метода класса FileOutputFormat для реализации сжатия:          /**
           * Устанавливает, сжаты ли выходные данные задания.
           * @param job задание для модификации
           * @param compress должно ли задание сжимать свои выходные данные?
           */
          public static void setCompressOutput(Job job, boolean compress) {
            job.getConfiguration().setBoolean(FileOutputFormat.COMPRESS, compress);
          }
          /**
           * Устанавливает {@link CompressionCodec}, используемый для сжатия выходных данных задания.
           * @param job задание для модификации
           * @param codecClass {@link CompressionCodec}, используемый для сжатия выходных данных задания
           */
          public static void setOutputCompressorClass(Job job, 
                                                      Class<? extends CompressionCodec> codecClass) {
            setCompressOutput(job, true);
            job.getConfiguration().setClass(FileOutputFormat.COMPRESS_CODEC, 
                                            codecClass, 
                                            CompressionCodec.class);
          }
          Если требуется, чтобы MapReduce выдавал последовательный файл (sequence file), можно настроить атрибут mapred.output.compression.type для управления конкретным типом сжатия. По умолчанию значение этого атрибута равно RECORD, что означает сжатие каждой отдельной записи. Рекомендуется установить значение типа на BLOCK, чтобы сжимать группу записей, что позволяет достичь более высокой эффективности сжатия. Также можно использовать статический метод putCompressionType() класса SequenceFileOutputFormat для настройки параметров сжатия.Хотя MapReduce напрямую обрабатывает исходные данные, но если можно сжать промежуточные результаты, полученные на этапе map, это всё равно повысит производительность программы. Промежуточные результаты, созданные на этапе map, должны быть записаны на диск и затем переданы через сеть на узел reducer. Если использовать быстрые методы сжатия, такие как LZO, LZ4 или Snappy, перед передачей данных через сеть, можно уменьшить объём передаваемых данных, что позволит повысить производительность. Если требуется, чтобы результаты map были сжаты в формат gzip, можно использовать следующий код:

```java
Configuration config = new Configuration();
config.setBoolean("mapred.compress.map.output", true);
config.setClass("mapred.map.output.compression.codec", GzipCodec.class, CompressionCodec.class);
Job job = new Job(config);