向 Hadoop/Spark 添加自定义代码——压缩编解码器

Question

在处理数据压缩时，Spark 支持底层 Hadoop 基础架构中的各种压缩方案。例如 Snappy（默认）、LZ4、LZF、GZIP。

如何指定使用与现有编解码器不同的用户构建的自定义编解码器顺序。例如，我的编解码器称为 DUMB。如何使用 DUMB 而不是默认的 Snappy。我查看了 CompressionCodecFactory 类 (https://hadoop.apache.org/docs/r1.2.1/api/org/apache/hadoop/io/compress/CompressionCodecFactory.html)，但仍然不太了解如何连接起来。以前有没有人做过类似的事情，或者有任何提示？

---

编辑：根据@Paweł_Jurczenko 的回答，我添加了更多细节。

这里是哑巴：

public class Dumb{

  public Dumb() {

  }

  public int CompressIt(InBuffers inBuffs) {
    return CallCompressor(inBuffs);
  } 
}

*InBuffers 是一个实现要压缩的缓冲区列表的类，而 CallCompressor 完成肮脏的地下工作。

要在主类中使用 DUMB，我会这样做：

Dumb myDumbComp = new Dumb();
myDumbComp.CompressIt(inBuffs)     //inBuffs is a List of individual input buffers

但是现在，我想要一个标准的 CompressionCodec 接口，通过它我可以调用 Hadoop 中的方法。

Answer 1

首先，您的自定义编解码器应该实现CompressionCodec 接口。然后您应该将spark.io.compression.codec 属性设置为您的编解码器的完全限定类名，例如：

val sparkConf: SparkConf = new SparkConf()
  .setAppName("...")
  .set("spark.io.compression.codec", "com.organization.compress.CustomCodec")
val sc: SparkContext = new SparkContext(sparkConf)

从现在开始，您的CustomCodec 将用于压缩内部 Spark 数据，如 RDD 分区、广播变量和随机输出。当然，您的编解码器必须存在于类路径中。如果您还想压缩输出数据，您应该设置hadoopConfiguration（SparkContext 的一部分）的以下属性：

sc.hadoopConfiguration.set("mapreduce.output.fileoutputformat.compress", "true")
sc.hadoopConfiguration.set("mapreduce.output.fileoutputformat.compress.codec", "com.organization.compress.CustomCodec")
sc.hadoopConfiguration.set("mapreduce.output.fileoutputformat.compress.type", "BLOCK")