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异常