pyspark 以减少/压缩的小文件数量写入配置单元表

Question

每次进程运行时我都会更新一个数据帧记录，这意味着每次进程完成时我都会有一个一行 4 列的数据帧。 然后我将使用 dataframe write 和 parquet 格式将它插入到 hive 表中。 由于一次只有一条记录，我在 hfds 的 table 文件夹中看到了很多小文件。

当我将数据写入 hive 表时，您能否告诉我如何减少并将其写入同一个文件（parquet 文件）？？

hdfs location: user_id/employe_db/market_table/
from:
part-04498-f33fc4b5-47d9-4d14-b37e-8f670cb2c53c-c000.snappy.parquet
part-04497-f33fc4b5-47d9-4d14-b37e-8f670cb2c53c-c000.snappy.parquet
part-04496-f33fc4b5-47d9-4d14-b37e-8f670cb2c53c-c000.snappy.parquet
part-04496-f33fc4b5-47d9-4d14-b37e-8f670cb2c53c-c000.snappy.parquet
part-04450-f33fc4b5-47d9-4d14-b37e-8f670cb2c53c-c000.snappy.parquet
part-04449-f33fc4b5-47d9-4d14-b37e-8f670cb2c53c-c000.snappy.parquet

to:
part-03049-f33fc4b5-47d9-4d14-b37e-8f670cb2c53c-c000.snappy.parquet

如何将镶木地板文件的数量减少到固定数量的更少文件并将新数据加载/写入现有文件？ part-04499-f33fc4b5-47d9-4d14-b37e-8f670cb2c53c-c000.snappy.parquet

Answer 1

在写信给 HDFS 之前，您可以 repartition(1) 以便每次执行创建 1 个文件。

df.repartition(1).write.parquet("<directory>")

Merging files:

Using Hive:

如果您已经在 user_id/employe_db/market_table/ 目录顶部有 hive 表，则通过选择同一个表运行插入覆盖。

spark.sql("insert overwrite table <db>.<tab_market> select * from <db>.<tab_market>")

--只创建一个文件然后使用order by

spark.sql("insert overwrite table <db>.<tab_market> select * from <db>.<tab_market> order by <column>")

您也可以像在 Hive 中一样运行插入语句。

（或）

Using Spark:

作为后期摄取过程，您可以再次从目录中读取 parquet 文件，然后再次重新分区并写入目录。

df_src=spark.read.parquet("<directory>")
df_src.repartition(<number>).write.mode("overwrite").parquet("<directory>")

NOTE

• 覆盖首先删除目录，以防万一在这之间作业失败，我们可能会遇到数据丢失。
• 最佳做法是将数据备份到 tmp 目录然后仅覆盖