spark读写压缩文件API使用详解

最近研究了下Spark如何读写压缩格式的文件，主要有如下三种方式，这里以lzo方式压缩为例

/*******************old hadoop api*************************/
val confHadoop = new JobConf
confHadoop.set("mapred.output.compress", "true")
confHadoop.set("mapred.output.compression.codec", "com.hadoop.compression.lzo.LzopCodec")
val textFile = sc.hadoopFile(args(0), classOf[DeprecatedLzoTextInputFormat],classOf[LongWritable], classOf[Text], 1)
textFile.saveAsHadoopFile(args(1),classOf[LongWritable], classOf[Text], classOf[TextOutputFormat[LongWritable,Text]],confHadoop)

/*******************new hadoop api*************************/
val job = new Job()
job.setOutputFormatClass(classOf[TextOutputFormat[LongWritable,Text]])
job.getConfiguration().set("mapred.output.compress", "true")
job.getConfiguration().set("mapred.output.compression.codec", "com.hadoop.compression.lzo.LzopCodec")
val textFile = sc.newAPIHadoopFile(args(0), classOf[LzoTextInputFormat],classOf[LongWritable], classOf[Text],job.getConfiguration())
textFile.saveAsNewAPIHadoopFile(args(1), classOf[LongWritable], classOf[Text],classOf[TextOutputFormat[LongWritable,Text]],job.getConfiguration())

/*******************textFile*************************/
val textFile = sc.textFile(args(0), 1)
textFile.saveAsTextFile(args(1),  classOf[LzopCodec])

以上三种方式，基本上利用到了所有Spark提供的主流读写文件的API，第一个案例是针对Spark提供的旧版Hadoop API来编写的，在JobConf中配置压缩的属性，读写的时候声明InputFormat和OutputFormat即可。第二个案例是采用新版Hadoop API的方式来编写的，步骤跟第一个案例类似。最后就是最简单的一种写法，在写入的时候指定Codec即可。

为了使Spark支持读写压缩格式的文件，还需要做一些基本的配置使得Spark可以加载需要用到的压缩格式相关的类库和jar，具体配置如下

spark.executor.extraLibraryPath=/usr/lib/native/
spark.executor.extraClassPath=/usr/lib/hadoop/lib/hadoop-lzo.jar

Spark支持三种配置属性的方式，优先级由低到高分别是：在conf/spark-defaults.conf中配置，使用spark-submit或spark-shell提交程序的时候配置参数，在Spark程序中通过System.setProperty方法或者通过设置SparkConf对象来配置参数。如果同一个参数配置了多次，以优先级最高的方式中配置的为准。
上述有关压缩的配置是针对Executor来说的，还需要在提交的时候针对Driver配置压缩相关的属性

--driver-class-path /usr/lib/hadoop/lib/hadoop-lzo.jar
--driver-library-path /usr/lib/native

在使用Spark SQL的时候，配置Executor和Driver与压缩相关的属性就可以正常读取Hive目录下的压缩文件(我测试的版本是CDH5.0.0和Spark1.0),如果想通过Hive计算后输出压缩格式的结果，则在hql()方法中设置压缩相关的属性即可，例如

hql("set io.compression.codecs=com.hadoop.compression.lzo.LzoCodec,com.hadoop.compression.lzo.LzopCodec")
hql("set io.compression.codec.lzo.class=com.hadoop.compression.lzo.LzoCodec")
hql("set mapred.output.compression.codec=com.hadoop.compression.lzo.LzopCodec")

如果NameNode配置了HA，则需要将hdfs-site.xml放入$SPARK_HOME/conf目录下，否则将导致读写HDFS的时候，无法加载到配置文件中的HA provider类，报UnknownHostException异常