场景描述

• 场景一：数据不均匀，个别task获取的数据比其他Task多，导致单个Task或几个Task执行很慢。
• 场景二：数据均匀，但是每个Task数据量都很多，执行时间达不到预期。

解决思路

• 读取前优化：优化文件存储，比如Hive分区存储，控制每个文件数据量等；
• 读取时优化：在读取hive或者文件时优化（推荐）；
• 读取后优化：在读取文件后进行优化。

常用手段

• 读取前优化：在文件读取前优化文件，比如Hive分区、分桶、压缩存储，控制每个文件数据量等；
• 读取时优化：在读取hive或者文件时优化（支持的文件格式：Parquet, JSON and ORC.）（推荐）；
  • 使用：spark.sql.files.maxPartitionBytes（非常有用），设置读取文件时每个分区的最大字节数，从而控制每个Task处理的数据量。通过减小这个参数的值，可以增加分区的数量，从而增加Task的数量。
  • 使用：spark.sql.files.openCostInBytes，设置打开文件的固定开销，影响文件分区的决策。在计算分区数时，Spark会考虑打开文件的成本。如果文件很小，Spark可能会将多个文件合并到同一个分区中，即使这意味着超过maxPartitionBytes的限制。在处理大量小文件时，调整这个参数可以帮助Spark更合理地分配Task。

spark.conf.set("spark.sql.files.maxPartitionBytes", (maxPartitionBytes * 1024 * 1024).toString) // 设置每个分区最大为128MB
spark.conf.set("spark.sql.files.openCostInBytes", "4194304") // 设置为4MB（这个也可以不设置）

• 读取后优化：在读取文件后进行优化。
  • 使用repartition或coalesce
  • 使用repartitionAndSortWithinPartitions
  • 使用spark.sql.shuffle.partitions

优化结果

通过上面的方式，基本控制每个Task在18MB左右，每个task3~6分钟左右，整体也就7分钟左右完成，之前基本需要30分钟左右