Apache Spark 的数据处理优化

本文介绍如何优化 Apache Spark 群集的配置，以在 Azure HDInsight 上实现最佳性能。

概述

如果在 Join 或 Shuffle 上出现作业执行缓慢的情况，原因可能是 数据倾斜。 数据倾斜指的是作业数据中的不对称分布。 例如，地图作业可能需要 20 秒。 但运行关联或重排序数据的作业需要数小时。 若要修复数据倾斜，应为整个密钥加盐，或者只对某些键子集使用 隔离盐 。 如果使用隔离的盐，则应进一步筛选以隔离映射联接中的加盐键子集。 另一种方法是引入一个存储桶列，并首先在存储桶中进行预聚合。

导致联接速度缓慢的另一个因素可能是联接类型。 默认情况下，Spark 使用 SortMerge 联接类型。 这种类型的联接最适合大型数据集。 但计算成本高昂，因为它必须先对数据的左右两侧进行排序，然后再合并数据。

Broadcast联接最适合较小的数据集，或者联接的一侧比另一端小得多。 这种类型的联接会将一侧的数据广播到所有执行程序，因此需要更多内存来进行整体广播。

可以通过设置 spark.sql.autoBroadcastJoinThreshold更改配置中的联接类型，也可以使用 DataFrame API （dataframe.join(broadcast(df2))） 设置联接提示。

// Option 1
spark.conf.set("spark.sql.autoBroadcastJoinThreshold", 1*1024*1024*1024)

// Option 2
val df1 = spark.table("FactTableA")
val df2 = spark.table("dimMP")
df1.join(broadcast(df2), Seq("PK")).
    createOrReplaceTempView("V_JOIN")

sql("SELECT col1, col2 FROM V_JOIN")

如果使用的是分桶表，则具有第三个联接类型，即 Merge 联接。 预先正确分区和排序的数据集将跳过SortMerge联接中的昂贵排序阶段。

联接顺序很重要，尤其是在更复杂的查询中。 从最有选择性的联接开始。 此外，移动联接，以尽可能增加聚合后的行数。

若要管理笛卡尔联接的并行度，可以添加嵌套结构、窗口化，并可能跳过 Spark 作业中的一个或多个步骤。

优化作业执行

• 根据需要进行缓存，例如，如果使用两次数据，则缓存数据。
• 将变量广播到所有执行程序。 变量仅序列化一次，从而加快查找速度。
• 在驱动程序上使用线程池，这会导致许多任务的操作速度更快。

定期监视正在运行的作业，以了解性能问题。 如果需要深入了解某些问题，请考虑以下性能分析工具之一：

• Intel PAL 工具 监视 CPU、存储和网络带宽使用情况。
• Oracle Java 8 监控 分析 Spark 和执行程序代码。

实现 Spark 2.x 查询性能的关键是 Tungsten 引擎，它依赖于整个阶段的代码生成。 在某些情况下，可能会禁用整个阶段代码生成。 例如，如果在聚合表达式string中使用非可变类型，则会显示SortAggregate而不是HashAggregate。 例如，为了获得更好的性能，请尝试以下操作，然后重新启用代码生成：

MAX(AMOUNT) -> MAX(cast(AMOUNT as DOUBLE))

后续步骤

• 优化 Apache Spark 的数据存储
• 优化 Apache Spark 的内存使用情况
• 优化 Apache Spark 的群集配置