<返回更多

Apache Spark优化工具包

2020-07-22    
加入收藏
Apache Spark优化工具包

> Source: Pixabay

 

Apache Spark是一种开放源代码的分布式计算引擎,目前是用于内存中批处理驱动的数据处理的最受欢迎的框架(它还支持实时数据流传输)。 得益于其先进的查询优化器,DAG调度程序和执行引擎,Spark能够非常高效地处理和分析大型数据集。 但是,在没有仔细调整的情况下运行Spark作业仍会导致性能下降。

在博客文章中,我将分享一些Spark性能调优的技巧,以帮助您解决和加快运行缓慢的Spark作业。

(本文提到的所有功能均来自PySpark,您可以使用Spark API文档找到与Scala / JAVA等效的功能。)

分区不均匀

当数据集最初由Spark加载并成为弹性分布式数据集(RDD)时,所有数据均匀地分布在分区之间。 但是,在用户对其应用某些类型的数据操作之后,这些分区可能会变得不均匀。 例如,groupByKey操作可能导致分区偏斜,因为一个键可能比另一个键包含更多的记录。 此外,由于Spark的DataFrameWriter允许使用partitionBy将分区数据写入磁盘,因此磁盘上的分区也可能不均匀。

在DataFrame中重新平衡偏斜的分区将极大地提高Spark在DataFrame上的处理性能。 您可以使用getNumPartitions函数检查DataFrame中的分区数,并通过运行简单的Spark作业来查找每个分区中的记录数,例如:

from pyspark.sql.functions import spark_partition_id
df.withColumn("partition_id", spark_partition_id()) 
  .groupBy("partition_id") 
  .count()
  .show()

如果发现DataFrame的分区大小高度不均匀,请在对它进行任何分析之前,使用重新分区或合并函数对DataFrame进行重新分区。 还建议在将数据写回磁盘之前,先对内存中的数据进行分区。 RDD模块也支持这些重新分区功能。

坚持RDD的缺点

由于惰性执行原理,除非用户明确调用操作来收集结果,否则Spark不会对数据集执行任何实际的转换。 此外,如果用户希望对中间结果应用其他转换,Spark将需要从头开始重新计算所有内容。 为了允许用户更有效地重用日期,Spark可以使用持久性或缓存功能将数据缓存在内存和/或磁盘中。

但是,缓存并不总是一个好主意。 Spark缓存数据集后,Catalyst优化器优化进一步转换的能力将受到限制,因为它不再能够改善源数据级别的修剪。 例如,如果将过滤器应用于在源数据库中建立索引的列,则Catalyst将无法利用索引来提高性能。

因此,仅当缓存数据将在以后多次重用时才建议使用缓存数据。 迭代探索数据集或调整ML模型时。

Apache Spark优化工具包

> Source: Pixabay

 

基于成本的优化器(CBO)

基于成本的优化器(CBO)通过向Catalyst提供其他表级统计信息,可以加快Spark SQL作业的速度,这对于连接许多数据集的作业特别有用。 使用者可以通过将spark.sql.cbo.enabled设置为true(默认值)来启用CBO。

为了充分利用CBO,用户需要保持列级和表级统计信息都是最新的,从而使CBO可以使用准确的估算来优化查询计划。 为此,在对表运行SQL查询之前,请使用ANALYZE TABLE命令收集统计信息。 记住在修改表之后再次分析表,以确保统计信息是最新的。

广播Join

除了启用CBO,在Spark中优化连接数据集的另一种方法是使用广播联接。 在无序连接中,两个表中的记录都将通过网络传输给执行器,当一个表比另一个大得多时,这是次优的。 在广播联接中,较小的表将被发送给执行程序,以与较大的表联接,从而避免了通过网络发送大量数据的情况。

用户可以通过spark.sql.autoBroadcastJoinThreshold配置控制广播联接,指示要广播的表的最大大小。 此外,即使表的大小大于spark.sql.autoBroadcastJoinThreshold,也可以使用广播提示来告诉Spark广播表:

from pyspark.sql.functions import broadcast
broadcast(spark.table("tbl_a")).join(spark.table("tbl_b"), "key")

垃圾收集(GC)

由于所有Spark作业都占用大量内存,因此确保有效进行垃圾收集非常重要-我们希望产生较少的内存"垃圾"以减少GC时间。 要了解您的Spark作业是否在GC中花费过多时间,请在Spark UI中检查"任务反序列化时间"和" GC时间"。

例如,由于Spark需要反序列化更多对象,因此使用用户定义函数(UDF)和lambda函数将导致更长的GC时间。 还建议避免创建中间对象并将不必要的RDD缓存到JVM堆。

TL; DR:

· 使用重新分区或合并来重新平衡不均匀的分区。

· 仅当数据将被多次重用时才保留数据。

· 使用ANALYZE TABLE命令可以维护CBO的最新统计信息。

· 为小表启用广播连接以加快连接速度。

· 通过使用较少的UDF并避免缓存大对象来优化GC。

(本文翻译自Xinran Waibel的文章《Apache Spark Optimization Toolkit》,参考:https://towardsdatascience.com/apache-spark-optimization-toolkit-17cf3e491992)

声明:本站部分内容来自互联网,如有版权侵犯或其他问题请与我们联系,我们将立即删除或处理。
▍最近更新
更多资讯 >>>