PySpark 数据从非分区 hive 表加载到分区 hive 表的性能调整答案

【问题标题】：Performance tuning for PySpark data load from a non-partitioned hive table to a partitioned hive tablePySpark 数据从非分区 hive 表加载到分区 hive 表的性能调整
【发布时间】：2021-06-22 08:54:18
【问题描述】：

我们需要从非分区EXTERNAL配置单元表work_db.customer_tbl中提取数据到通过 PySpark 的分区 EXTERNAL hive 表 final_db.customer_tbl，之前通过 hive 查询完成。最终表由 load_date 列进行分区（load_date 列的格式为 yyyy-MM-dd）。

所以我们有一个简单的 PySpark 脚本，它使用插入查询（与之前使用的 hive 查询相同），使用 spark.sql()获取数据> 命令。但是我们有一些严重的性能问题，因为我们在摄取后尝试摄取的表有大约 3000 个分区，每个分区都有大约 4 MB 的数据，除了最后一个分区大约是4GB。总表大小接近 15GB。此外，在摄取后，每个分区都有 217 个文件。决赛桌是一张简洁的压缩拼花桌。

源工作表有一个 15 GB 文件，文件名格式为 customers_tbl_unload.dat。

早些时候，当我们通过直线连接使用 hive 查询时，通常需要大约 25-30 分钟才能完成。现在，当我们尝试使用 PySpark 脚本时，大约需要 3 个小时才能完成。

我们如何调整火花性能以使摄取时间少于直线所需的时间。

The configurations of the yarn queue we use is:
Used Resources: <memory:5117184, vCores:627>
Demand Resources:   <memory:5120000, vCores:1000>
AM Used Resources:  <memory:163072, vCores:45>
AM Max Resources:   <memory:2560000, vCores:500>
Num Active Applications:    45
Num Pending Applications:   45
Min Resources:  <memory:0, vCores:0>
Max Resources:  <memory:5120000, vCores:1000>
Reserved Resources: <memory:0, vCores:0>
Max Running Applications:   200
Steady Fair Share:  <memory:5120000, vCores:474>
Instantaneous Fair Share:   <memory:5120000, vCores:1000>
Preemptable:    true

The parameters passed to the PySpark script is:
num-executors=50
executor-cores=5
executor-memory=10GB

PySpark code used:
insert_stmt = """INSERT INTO final_db.customers_tbl PARTITION(load_date) 
SELECT col_1,col_2,...,load_date FROM work_db.customer_tbl"""
spark.sql(insert_stmt)

即使在几乎使用了纱线队列的 10% 资源之后，这项工作也花费了很多时间。我们如何调整工作以提高效率。

【问题讨论】：

标签： apache-spark pyspark hive apache-spark-sql

【解决方案1】：

您需要重新分析您的数据集，并通过在日期列上对 yoir 数据集进行分区来查看您是否使用了正确的方法，或者您是否应该按年份进行分区？要了解为什么每个分区最终会有 200 多个文件，您需要了解 Spark 和 Hive 分区之间的区别。您应该首先尝试的直接方法是将输入数据集作为数据帧读取，并按您计划在 Hive 中用作分区键的键对其进行分区，然后使用 df.write.partitionBy 保存它

由于日期列上的数据似乎也有偏差，请尝试将其分区到可能具有相同数据分布的其他列上。否则，过滤掉倾斜的数据，单独处理

【讨论】：