从spark 2.4开始，您可以在查询中添加一个提示来合并和重新划分最终的Select。

INSERT OVERWRITE TABLE ... PARTITION (...) 
SELECT /*+ REPARTITION(5) */ client_id, country FROM mytable;

这将生成5个文件。
在Spark 2.4之前，可能会影响查询性能，您可以将spark.sql.shuffle.partitions设置为所需文件的数量。

（我知道我们可能应该使用DISTRIBUTE BY分区列，但这只是提供了一个最简单的示例）
所以看起来我简化的尝试是我出错的地方。如果我DISTRIBUTE BY实际的列而不是人为的1（即DISTRIBUTE BY load_date或其他），那么它不会创建空文件。为什么？谁知道呢...
(This也匹配merge-multiple-small-files-in-to-few-larger-files-in-spark线程上的this答案）

这对我来说是一个真实的烦人的问题，我花了一段时间才解决。
以下两种方法对我很有效：
1.在外部将其作为直线脚本运行：

set hive.exec.dynamic.partition.mode=nonstrict;
     set hive.merge.mapfiles=true;
     set hive.merge.mapredfiles=true;
     set hive.merge.smallfiles.avgsize=64512000;
     set hive.merge.size.per.task=12992400;
     set hive.exec.max.dynamic.partitions=2048;
     set hive.exec.max.dynamic.partitions.pernode=1024;

     <insert overwrite command>

这种方法的问题是，不知何故，它不能在pyspark内部工作，而可以作为外部beeline脚本从python脚本中运行
1.使用重新分区
我发现这个选项相当不错。Repartition（x）允许将pyspark Dataframe 的记录压缩到“x”文件中。
现在，由于不可能用一个静态的数字“x”来重新划分每个表，因为表的大小会变化（例如，我不想将一个有10 mil记录的表重新划分为1），我做了以下操作。

-> set an upper threshold for the max number of records a partition should hold 
I use 100,000 

-> compute x as : 
x = df.count()/max_num_records_per_partition
In case the table is partitioned,I use df_partition instead of df...i.e for every set of partition values, i filter df_partition from df; and then compute x from df_partition

-> repartition as:
df = df.repartition(x)
In case if the table is partitioned; i use df_partition = df_partition.repartition(x)

-> insert overwrite dataframe

这种方法对我来说很方便。
更进一步，表中的列数和用于每列的数据类型可用于创建权重，该权重可用于获得对给定dataFrame的再划分数的有效得多的估计（例如，与具有相同类型的5列的dataFrame相比，具有20列的dataFrame将获得更高的权重;与类型为Boolean的1列dataFrame相比，类型为Map的1列dataFrame将获得更高的权重

双向

合并比重新划分好。
阅读此主题：https://kontext.tech/article/1155/use-spark-sql-partitioning-hints

INSERT OVERWRITE TABLE ... PARTITION (...) 
SELECT /*+ REPARTITION(5) */ client_id, country FROM mytable;

INSERT OVERWRITE TABLE ... PARTITION (...) 
SELECT /*+ COALESCE(5) */ client_id, country FROM mytable;