在生产中，我们往往将Spark SQL作为Hive的替代方案，以获得SQL on Hadoop更出色的性能。因此，本文所讲的是指存储于HDFS中小文件，即指文件的大小远小于HDFS上块（dfs.block.size）大小的文件。
一方面，大量的小文件会给Hadoop集群的扩展性和性能带来严重的影响。NameNode在内存中维护整个文件系统的元数据镜像，用户HDFS的管理；其中每个HDFS文件元信息（位置，大小，分块等）对象约占150字节，如果小文件过多，会占用大量内存，直接影响NameNode的性能。相对的，HDFS读写小文件也会更加耗时，因为每次都需要从NameNode获取元信息，并与对应的DataNode建立连接。如果NameNode在宕机中恢复，也需要更多的时间从元数据文件中加载。
另一方面，也会给Spark SQL等查询引擎造成查询性能的损耗，大量的数据分片信息以及对应产生的Task元信息也会给Spark Driver的内存造成压力，带来单点问题。此外，入库操作最后的commit job操作，在Spark Driver端单点做，很容易出现单点的性能问题。
1.数据源本身就含大量小文件。
2.动态分区插入数据。
没有Shuffle的情况下，输入端有多少个逻辑分片，对应的HadoopRDD就会产生多少个HadoopPartition，每个Partition对应于Spark作业的Task（个数为M），分区数为N。最好的情况就是（M=N）&（M中的数据也是根据N来预先打散的），那就刚好写N个文件；最差的情况下，每个Task中都有各个分区的记录，那文件数最终文件数将达到M * N个。这种情况下是极易产生小文件的。
比如我们拿TPCDS测试集中的store_sales进行举例， sql如下所示：
use tpcds_1t_parquet;INSERT overwrite table store_sales partition ( ss_sold_date_sk ) SELECT * FROM tpcds_1t_ext.et_store_sales;
首先我们得
到其执行计划，如下所示，
== Physical Plan ==InsertIntoHiveTable MetastoreRelation tpcds_1t_parquet, store_sales, Map(ss_sold_date_sk -> None), true, false+- HiveTableScan [ss_sold_time_sk
store_sales的原生文件包含1616逻辑分片，对应生成1616 个Spark Task，插入动态分区表之后生成1824个数据分区加一个NULL值的分区，每个分区下都有可能生成1616个文件，这种情况下，最终的文件数量极有可能达到2949200。
1T的测试集store_sales也就大概300g，这种情况每个文件可能就零点几M。
动态分区插入数据，有Shuffle的情况下，上面的M值就变成了spark.sql.shuffle.partitions（
默认值200
）这个参数值，文件数的算法和范围和2中基本一致。
比如，
为
了防止Shuffle阶段的数据倾斜我们可以在上面的sql中加上 distribute by rand()，这样我们的执行计划就变成了：
InsertIntoHiveTable MetastoreRelation tpcds_1t_parquet, store_sales, Map(ss_sold_date_sk -> None), true, false+- *Project [ss_sold_time_sk +- Exchange(coordinator id: 1080882047) hashpartitioning(_nondeterministic +- *Project [ss_sold_time_sk +- HiveTableScan [ss_sold_date_sk
这种情况下，这样我们的文件数妥妥的就是spark.sql.shuffle.partitions * N，因为rand函数一般会把数据打散的非常均匀。
当spark.sql.shuffle.partitions设置过大时，小文件问题就产生了；
当spark.sql.shuffle.partitions设置过小时，任务的并行度就下降了，性能随之受到影响。
最理想的情况当然是根据分区字段进行shuffle，在上面的sql中加上distribute by ss_sold_date_sk。把同一分区的记录都哈希到同一个分区中去，由一个Spark的Task进行写入，这样的话只会产生N个文件，在我们的case中store_sales，在1825个分区下各种生成了一个数据文件。
但是这种情况下也容易出现数据倾斜的问题，比如双11的销售数据就很容易在这种情况下发生倾斜。
（基于分区字段Shuffle可能出现数据倾斜）
如上图所示，在我们插入store_sales时，就发生了null值的倾斜，大大的拖慢的数据入库的时间。
前面已经提到根据分区字段进行分区，除非每个分区下本身的数据较少，分区字段选择不合理，那么小文件问题基本上就不存在了，但也有可能由于shuffle引入新的数据倾斜问题。
首先可以尝试是否可以将两者结合使用， 在之前的sql上加上distribute by ss_sold_date_sk，cast(rand() * 5 as int)， 这个类似于我们处理数据倾斜问题时候给字段加上后缀的形式。
use tpcds_1t_parquet;INSERT overwrite table store_sales partition ( ss_sold_date_sk ) SELECT * FROM tpcds_1t_ext.et_store_salesdistribute by ss_sold_date_sk, cast(rand() * 5 as int);
按照之前的推算，每个分区下将产生5个文件，同时null值倾斜部分的数据也被打散成五份进行计算，缓解了数据倾斜的问题 ，我们最终将得到1825 *5=9105个文件，如下所示
1825 9105 247111074494 /user/kyuubi/hive_db/tpcds_1t_parquet.db/store_sales
如果我们将5改
得更小，文件数也会越少，但相应的倾斜key的计算时间也会上去。
在我们知道那个分区键倾斜的情况下，我们也可以将入库的SQL拆成几个部分，比如我们store_sales是因为null值倾斜，我们就可以通过where ss_sold_date_sk is not null 和 where ss_sold_date_sk is null 将原始数据分成两个部分。前者可以基于分区字段进行分区，如distribute by ss_sold_date_sk；后者可以基于随机值进行分区，distribute by cast(rand() * 5 as int)，这样可以静态的将null值部分分成五个文件。
FROM tpcds_1t_ext.et_store_sales where ss_sold_date_sk is not nulldistribute by ss_sold_date_sk;
FROM tpcds_1t_ext.et_store_sales where ss_sold_date_sk is nulldistribute by distribute by cast(rand() * 5 as int);
对于倾斜部分的数据，我们可以开启Spark SQL的自适应功能，spark.sql.adaptive.enabled=true来动态调整每个相当于Spark的reduce端task处理的数据量，这样我们就不需要人为的感知随机值的规模了，我们可以直接
FROM tpcds_1t_ext.et_store_sales where ss_sold_date_sk is nulldistribute by distribute by rand() ;
然后Spark在Shuffle 阶段会自动的帮我们将数据尽量的合并成spark.sql.adaptive.shuffle.targetPostShuffleInputSize（默认64m）的大小，以减少输出端写文件线程的总量，最后减少个数。
对于spark.sql.adaptive.shuffle.targetPostShuffleInputSize参数而言，我们也可以设置成为dfs.block.size的大小，这样可以做到和块对齐，文件大小可以设置的最为合理。
在我们的网易有数大数据平台上面，随便的建立几个SQL作业，不用会Spark也可以用SQL把大数据玩得666！
（网易有数大数据平台 - 数据开发）
从左到右依次为
-
建表 - 按分区字段插入非空集合到分区表 - 按rand插入空集到分区表，并开启自Spark SQL适应；
-
建表 - 不shuffle 按原始分片直接插入分区表；
-
建表 - 全集按照分区字段插入到分区表；
-
建表 - 全局按分区字段+cast(rand() * 5 as int)方式插入分区表。
双击每个工作节点，我们也可以对我们的SQL作业进行参数的调整
（网易有数大数据平台
- 参数配置）
选中我们对应的实验组，点击执行后，可以查看任务的运行状态。
（网易有数大数据平台
- 运行状态）
从各组的实验结果来看
bin/hadoop fs -count /user/kyuubi/hive_db/tpcds_1t_parquet.db/store_sales/1825 1863 192985051585 /user/kyuubi/hive_db/tpcds_1t_parquet.db/store_sales
bin/hadoop fs -du -h /user/kyuubi/hive_db/tpcds_1t_parquet.db/store_sales/ss_sold_date_sk=__HIVE_DEFAULT_PARTITION__183.1 M /user/kyuubi/hive_db/tpcds_1t_parquet.db/store_sales/ss_sold_date_sk=__HIVE_DEFAULT_PARTITION__/part-00000183.1 M /user/kyuubi/hive_db/tpcds_1t_parquet.db/store_sales/ss_sold_date_sk=__HIVE_DEFAULT_PARTITION__/part-00001183.3 M /user/kyuubi/hive_db/tpcds_1t_parquet.db/store_sales/ss_sold_date_sk=__HIVE_DEFAULT_PARTITION__/part-00002183.1 M /user/kyuubi/hive_db/tpcds_1t_parquet.db/store_sales/ss_sold_date_sk=__HIVE_DEFAULT_PARTITION__/part-00003183.1 M /user/kyuubi/hive_db/tpcds_1t_parquet.db/store_sales/ss_sold_date_sk=__HIVE_DEFAULT_PARTITION__/part-00004183.2 M /user/kyuubi/hive_db/tpcds_1t_parquet.db/store_sales/ss_sold_date_sk=__HIVE_DEFAULT_PARTITION__/part-00005183.2 M /user/kyuubi/hive_db/tpcds_1t_parquet.db/store_sales/ss_sold_date_sk=__HIVE_DEFAULT_PARTITION__/part-00006......
bin/hadoop fs -du -h /user/kyuubi/hive_db/tpcds_1t_parquet.db/store_sales/ss_sold_date_sk=2452642194.5 M /user/kyuubi/hive_db/tpcds_1t_parquet.db/store_sales/ss_sold_date_sk=2452642/part-00369
实验组一的小文件控制还是可喜可贺的。对于我们1t的tpcds测试数据，null值分区字段下只有40个文件，其他每个数据分区也只有一个数据文件，总目录1825，总文件数1863. 在解决数据倾斜问题的基础上，也只比纯按照分区字段进行distibute by多了39个文件。
本文讲述的是如何在纯写SQL的场景下，如何用Spark SQL做数据导入时候，控制小文件的数量。
-
对于原始数据进行按照分区字段进行shuffle，可以规避小文件问题。但有可能引入数据倾斜的问题；
