由于似乎没有人知道我的问题的答案，我想与大家分享我们采取的方法，使这个处理更有效，评论是非常欢迎的。
我们发现了（医生。不太清楚）hdfs中spark“checkpoints”中存储的一些信息是一些元数据文件，描述了每个Parquet地板文件的写入时间和大小：

$hdfs dfs -ls -h hdfs://...../my_spark_job/_spark_metadata

w-r--r--   3 hdfs 68K   2020-02-26 20:49 hdfs://...../my_spark_job/_spark_metadata/3248
rw-r--r--  3 hdfs 33.3M 2020-02-26 20:53 hdfs://...../my_spark_job/_spark_metadata/3249.compact
w-r--r--   3 hdfs 68K   2020-02-26 20:54 hdfs://...../my_spark_job/_spark_metadata/3250
...

$hdfs dfs -cat hdfs://...../my_spark_job/_spark_metadata/3250
v1
{"path":"hdfs://.../my_spark_job/../part-00004.c000.snappy.parquet","size":9866555,"isDir":false,"modificationTime":1582750862638,"blockReplication":3,"blockSize":134217728,"action":"add"}
{"path":"hdfs://.../my_spark_job/../part-00004.c001.snappy.parquet","size":526513,"isDir":false,"modificationTime":1582750862834,"blockReplication":3,"blockSize":134217728,"action":"add"}
...

所以，我们做的是：
构建一个spark流作业 _spark_metadata 文件夹。
我们使用 fileStream 因为它允许我们定义要使用的文件过滤器。
该流中的每个条目都是这些json行中的一行，通过解析来提取文件路径和大小。
按文件所属的父文件夹（Map到每个impala分区）对文件进行分组。
对于每个文件夹：
读取仅加载目标Parquet文件的Dataframe（以避免与写入文件的其他作业发生竞争）
计算要写入的块数（使用json中的size字段和目标块大小）
将Dataframe合并到所需的分区数并将其写回hdfs
执行ddl REFRESH TABLE myTable PARTITION ([partition keys derived from the new folder] 最后，删除源文件
我们取得的成就是：
通过对每个分区和批执行一次刷新来限制DDL。
通过配置批处理时间和块大小，我们能够使产品适应具有较大或较小数据集的不同部署场景。
这个解决方案非常灵活，因为我们可以为spark流作业分配更多或更少的资源（执行器、内核、内存等），也可以启动/停止它（使用它自己的检查点系统）。
我们还研究了在执行此过程时应用一些数据重新分区的可能性，以使分区尽可能接近最佳大小。