spark df 가 hive 표를 삽입 한 후 작은 파일 의 수량 이 많 습 니 다. 어떻게 합병 합 니까?

원본 주소:https://www.cnblogs.com/yy3b2007com/p/7113936.html　
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 방식 으로 압축 하고 합병 합 니 다.
 

 1 set hive.exec.reducers.bytes.per.reducer=500000000;
 2 set hive.mapred.supports.subdirectories=true;
 3 set mapreduce.input.fileinputformat.input.dir.recursive=true;
 4 set mapred.max.split.size=256000000;
 5 set mapred.min.split.size.per.node=128000000;
 6 set mapred.min.split.size.per.rack=128000000;
 7 set hive.hadoop.supports.splittable.combineinputformat=true;
 8 set hive.input.format=org.apache.hadoop.hive.ql.io.CombineHiveInputFormat;
 9 set hive.merge.mapfiles=true;
10 set hive.merge.mapredfiles=true;
11 set hive.merge.size.per.task=256000000;
12 set hive.merge.smallfiles.avgsize=256000000;
13 set hive.groupby.skewindata=true;
14 set hive.exec.dynamic.partition.mode=nonstrict;
15 set hive.exec.parallel=true;
16 set hive.exec.parallel.thread.number=32;
17 SET hive.exec.compress.output=true;
18 SET mapred.output.compression.codec=org.apache.hadoop.io.compress.SnappyCodec;
19 SET mapred.output.compression.type=BLOCK;
20 set hive.exec.compress.intermediate=true;
21 set hive.intermediate.compression.codec=org.apache.hadoop.io.compress.SnappyCodec;
22 set hive.intermediate.compression.type=BLOCK;