在做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;
使用代码进行重新分区让其合并再写入:
val aDF =hiveContext.table("info_user").repartition(2).persist()
aDF.registerTempTable("info_user")
这里也可以是从是hive sql 中转到另外一张表,让后再写入到目标表的方式。