spark中的性能调优

Question

我正在运行一个处理大约 2 TB 数据的 spark 作业。处理涉及：

1. 读取数据（avrò 文件）
2. 在属于地图类型的列上展开
3. OrderBy 分解列中的键
4. 过滤DataFrame（我有一个非常小的（7）组键（称为键集），我想过滤df）。我做一个df.filter(col("key").isin(keyset: _*) )
5. 我将此 df 写入镶木地板（此数据框非常小）
6. 然后我再次过滤原始数据帧以获取不在键集中的所有键 df.filter(!col("key").isin(keyset: _*) ) 并将其写入镶木地板。这是更大的数据集。

原始的 avro 数据约为 2TB。处理大约需要 1 小时。我想优化它。我在第 3 步之后缓存数据帧，使用 6000 的随机分区大小。min executors = 1000，max = 2000，executor memory = 20 G，executor core = 2。还有其他优化建议吗？左连接会比过滤器性能更好吗？

Answer 1

在我看来一切都很好。 如果你有小数据集，那么isin 没问题。

1) 确保可以增加核心数。执行器核心=5

不建议每个执行器使用超过 5 个内核。这是基于一项研究，其中任何具有 5 个以上并发线程的应用程序都会开始影响性能。

2) 确保您有良好/统一的分区结构。

示例（仅用于调试目的，不用于生产）：

  import org.apache.spark.sql.functions.spark_partition_id
  yourcacheddataframe.groupBy(spark_partition_id).count.show()

这将打印火花分区号和多少条记录 存在于每个分区中。基于此，如果您不想更多并行，您可以重新分区。

3) spark.dynamicAllocation.enabled 可能是另一种选择。

例如：

spark-submit --conf spark.dynamicAllocation.enabled=true --conf spark.dynamicAllocation.cachedExecutorIdleTimeout=100 --conf spark.shuffle.service.enabled=true

以及所有其他必需的道具.....那是该工作的。如果您在 spark-default.conf 中提供这些道具，它将适用于所有工作。

使用上述所有这些选项，您的处理时间可能会缩短。

Answer 2

除上述内容外，根据您的要求和集群提出一些建议：

1. 如果作业可以在 20g 执行器内存和 5 个核心上运行，则可以通过减少执行器内存并保持 5 个核心来容纳更多工作人员
2. orderBy 真的是必需的吗？ Spark 确保行在分区内排序，而不是在分区之间排序，这通常不是非常有用。
3. 文件是否需要位于特定位置？如果没有，添加一个

df.withColumn("in_keyset", when( col('key').isin(keyset), lit(1)).otherwise(lit(0)). \
write.partitionBy("in_keyset").parquet(...)

可能会加快操作以防止数据被读取+爆炸2x。 partitionBy 确保键集中的项目与其他键位于不同的目录中。

Answer 3

spark.dynamicAllocation.enabled 已启用

分区大小非常不均匀（基于输出 parquet 部分文件的大小），因为我正在执行 orderBy 键并且某些键比其他键更频繁。

keyset 是一个非常小的集合（7 个元素）