Spark:spark df插入hive表后小文件數量多，如何合並？

　　在做spark開發過程中，時不時的就有可能遇到租戶的hive庫目錄下的文件個數超出了最大限制問題。

• 一般情況下通過hive的參數設置：

    val conf = new SparkConf().setAppName("MySparkJob")
    //.setMaster("local[1]").setMaster("spark://172.21.7.10:7077").setJars(List("xxx.jar")).set("spark.executor.memory", "10g")
    val sc = new SparkContext(conf)
    val hiveContext = new HiveContext(sc)

    hiveContext.sql("use myhivedb")

    // toDF() method need this line...
    import hiveContext.implicits._
    hiveContext.sql("set hive.mapred.supports.subdirectories=true")
    hiveContext.sql("set mapreduce.input.fileinputformat.input.dir.recursive=true")
    hiveContext.sql("set mapred.max.split.size=256000000")
    hiveContext.sql("set mapred.min.split.size.per.node=128000000")
    hiveContext.sql("set mapred.min.split.size.per.rack=128000000")
    hiveContext.sql("set hive.hadoop.supports.splittable.combineinputformat=true")
    hiveContext.sql("set hive.exec.compress.output=true")
    hiveContext.sql("set mapred.output.compression.codec=org.apache.hadoop.io.compress.GzipCodec")
    hiveContext.sql("set hive.input.format=org.apache.hadoop.hive.ql.io.CombineHiveInputFormat")

    hiveContext.sql("set hive.merge.mapfiles=true")
    hiveContext.sql("set hive.merge.mapredfiles=true")
    hiveContext.sql("set hive.merge.size.per.task=256000000")
    hiveContext.sql("set hive.merge.smallfiles.avgsize=256000000")

    hiveContext.sql("set hive.groupby.skewindata=true")

• 通過df.repartition(xxx).persist()來實現小文件合並

但是並不是所有的小文件都會合並的，例如：

val my_df = my_temp1_df.unionAll(my_temp2_df)
my_df.registerTempTable("temp_temphivetable")
hiveContext.sql("insert into temphivetable select * from temp_temphivetable")
// -rwxrwx---+ 3 jc_rc jc_rc_group 187.9 K 2017-06-28 17:58 /my tenant name/myhivedb/temphivetable/part-37944.gz
// -rwxrwx---+ 3 jc_rc jc_rc_group 188.9 K 2017-06-28 17:56 /my tenant name/myhivedb/temphivetable/part-37945.gz

當設置：repartition(1000)時

    // 當設置：repartition(1000)時，
    // -rwxrwx---+  3 jc_rc jc_rc_group     10.9 M 2017-07-01 16:31 /my tenant name/myhivedb/temphivetable/part-00998.gz
    // -rwxrwx---+  3 jc_rc jc_rc_group     10.9 M 2017-07-01 16:31 /my tenant name/myhivedb/temphivetable/part-00999.gz
    val my_df = my_temp1_df.unionAll(my_temp2_df).repartition(1000).persist()
    my_df.registerTempTable("temp_temphivetable")
    hiveContext.sql("insert into temphivetable select * from temp_temphivetable")

 當設置：repartition(100)時

    // 當設置：repartition(100)時，
    // -rwxrwx---+  3 jc_rc jc_rc_group    103.0 M 2017-07-01 17:53 /my tenant name/myhivedb/temphivetable/part-00098.gz
    // -rwxrwx---+  3 jc_rc jc_rc_group    103.2 M 2017-07-01 17:53 /my tenant name/myhivedb/temphivetable/part-00099.gz
    val my_df = my_temp1_df.unionAll(my_temp2_df).repartition(1000).persist()
    my_df.registerTempTable("temp_temphivetable")
    hiveContext.sql("insert into temphivetable select * from temp_temphivetable")

hivesql下采用snappy方式壓縮並且合並：

set hive.exec.reducers.bytes.per.reducer=500000000;
set hive.mapred.supports.subdirectories=true;
set mapreduce.input.fileinputformat.input.dir.recursive=true;
set mapred.max.split.size=256000000;
set mapred.min.split.size.per.node=128000000;
set mapred.min.split.size.per.rack=128000000;
set hive.hadoop.supports.splittable.combineinputformat=true;
set hive.input.format=org.apache.hadoop.hive.ql.io.CombineHiveInputFormat;
set hive.merge.mapfiles=true;
set hive.merge.mapredfiles=true;
set hive.merge.size.per.task=256000000;
set hive.merge.smallfiles.avgsize=256000000;
set hive.groupby.skewindata=true;
set hive.exec.dynamic.partition.mode=nonstrict;
set hive.exec.parallel=true;
set hive.exec.parallel.thread.number=32;
SET hive.exec.compress.output=true;
SET mapred.output.compression.codec=org.apache.hadoop.io.compress.SnappyCodec;
SET mapred.output.compression.type=BLOCK;
set hive.exec.compress.intermediate=true;
set hive.intermediate.compression.codec=org.apache.hadoop.io.compress.SnappyCodec;
set hive.intermediate.compression.type=BLOCK;

 使用代碼進行重新分區讓其合並再寫入：

    val aDF =hiveContext.table("info_user").repartition(2).persist()
 
    aDF.registerTempTable("info_user")

這里也可以是從是hive sql 中轉到另外一張表，讓后再寫入到目標表的方式。