【译】深入分析Spark UDF的性能-阿里云开发者社区

开发者社区> Apache Spark中国社区> 正文

【译】深入分析Spark UDF的性能

简介: 这篇博客会阐述一份关于Apache Spark的在Scala UDF、 PySpark UDF 和PySpark Pandas UDF之间的性能评测报告。

原文链接 https://medium.com/@QuantumBlack/spark-udf-deep-insights-in-performance-f0a95a4d8c62


编译:抚月,阿里巴巴计算平台事业部 EMR 高级工程师,Apache HDFS Committer,目前从事开源大数据存储和优化方面的工作。


这篇博客会阐述一份关于Apache Spark的在Scala UDF、 PySpark UDF 和PySpark Pandas UDF之间的性能评测报告。


Spark提供了多种解决方案来应对复杂挑战, 但是我们面临了很多场景, 原生的函数不足以解决问题。因此,Spark允许我们注册自定义函数(User-Defined Functions, 或者叫 UDFs)

在这片文章, 我们会探索Spark的UDF的性能特点。

Spark支持多种语言,比如Python, Scala, Java, R, SQL. 但是通常数据操作都是用PySpark或者Spark Scala写的。 我们认为Pyspark被大多数用户采用, 是因为以下原因:

  1. 更快的学习曲线  -- Python比Scala更简单。
  2. 更广的社区支持  -- 程序员对Pyspark性能等建议,反馈到社区,形成更好的生态。
  3. 丰富的可用的类库 -- Python有很多机器学习、时序分析、数理统计的类库。
  4. 很小的性能差异 -- Spark DataFrames引入之后,意味着Scala和Python的性能几乎相同。 Datafarme现在是按照带名字的列(named columns)来组织的, 这样Spark可以更好地理解Schema。而那些用来构建dataframe的操作,会被Catalyst Optimizer编译成物理执行计划(physical execution plan)来加速计算。
  5. 数据工程师和数据科学家,交接代码也更简单。有一些dataframe的操作需要UDFs, PySpark可能会有性能问题。有一些解决办法,就是将PySpark和Scala UDF, UDF Wrapper一起使用。

PySpark作业提交的时候, driver端跑在Python上, driver会创建一个SparkSession对象以及Dataframes/RDDs. 这些Python对象是一些wrapper对象, 本质是JVM(Java)对象。 为了简化,PySpark提供了一个wrapper来跑原生Scala代码。

Spark UDF 函数

通过Scala, Python 或者 Java 注册自定义函数,是非常通用的方法, 来扩展SQL用户的能力, 是的用户可以调用这些函数而不需要再写代码。

例如, 将一个100w行的集合乘以1000:

def times1000(field):
  return field * 1000.00

或者, 对经纬度数据集进行反向地理编码(reverse geocode):

import geohash
def geohash_pyspark(lat, lon):
  return geohash.encode(lat, lon)

Spark SQL提供了一种方法, 你可以用自己的编程语言来传入1个函数,从而注册UDF。Scala和Python可以用原生的函数或者lamdba语法,除了Java繁琐一些,需要扩展这个UDF类。
UDF可以作用于多种不同的数据类型,并返回一种不同的类型。在Python和Java里,我们需要指定发返回类型。

UDF可以通过以下方式进行注册:

spark.udf.register("UDF_Name", function_name, returnType())

*returnType() 在Python和Java里是强制的。

多种Spark UDF和执行方式

在分布式模式下,Spark使用master/worker架构来执行。调度器(driver)来跟大量的workers(或者叫executors)进行通信。 driver和worker跑在自己的Java进程里。
driver端通过main()方法,创建了SparkContext, RDDs并执行一些变换操作。 Executors负责跑一个个的任务。

Screen Shot 2019-12-14 at 20.40.53.png
1576224390592-89e2031e-a0f6-4acd-9b4b-58d5d667f264.png

性能基准测试

我们创建了一个随机的经纬度数据集, 包含100w条记录, 共1.2GB,来测试3种Spark UDF类型的性能。我们创建了2个UDF:一个简单的乘以1000的函数, 一个复杂的geohash函数。 (所以总共有2 * 3 = 6组测试)

集群配置: 8个节点
Driver节点: 16核 122GB内存
Worker节点: 4核 30.5GB内存,开启自动扩容
Notebook代码:  https://bit.ly/2YxiVp4 使用 QuantumBlack’s的方法来跑 Scala UDF, PySpark UDF and PySpark Pandas UDF 的测试。

除了上面3种类型的UDF,我们还创建了Python wrapper, 从而在Pyspark中调用Scala UDF。我们发现这种方式, 既可以使用简单的python编程,又能兼顾Scala UDF的性能。

用Pyspark代码来创建一个Python wrapper:

from pyspark.sql.column import Column
from pyspark.sql.column import _to_java_column
from pyspark.sql.column import _to_seq
from pyspark.sql.functions import col

def udfGeohashScalaWrapper(lat, lon):
    _geohash = sc._jvm.sparkudfperformance.UDFs.udfGeohash()
    return Column(_geohash.apply(_to_seq(sc, [lat, lon], _to_java_column)))
def udfTimes1000ScalaWrapper(field):
    _times1000 = sc._jvm.sparkudfperformance.UDFs.udfTimes1000()
    return Column(_times1000.apply(_to_seq(sc, [field], _to_java_column)))

Databricks对 Pandas UDF 做过一份性能报告  https://databricks.com/blog/2017/10/30/introducing-vectorized-udfs-for-pyspark.html

重要结论

下面是测试结果

1576224428174-0d814c3d-160e-4ffa-8551-b4b5705c149b.png


测试结果中, Scala UDF的性能是最好的。 前面提到, Scala和Python之间的转换步骤, 使得Python UDF需要处理更多东西。
我们同时发现,PySpark Pandas UDF在小数据集或者简单函数上,性能好于PySpark UDF。而如果是一个复杂的函数,比如引入了geohash,这种场景下 PySpark UDF的性能会比PySpark Pandas UDF好10倍。
我们还发现了,在PySpark代码里, 创建一个Python wrapper去调用Scala UDF,性能比这两种PySpark UDFs好15倍。


综合考虑了上面的一些性能特征, QuantumBlack公司现在采用的方式是:

  • 使用 PySpark UDF, 如果数据集不大,并且需要用简单函数进行快速的数据洞察。
  • 构架一个可复用的Scala UDF的内置库。
  • 创建Python wrapper来调用Scala UDF

版权声明:本文中所有内容均属于阿里云开发者社区所有,任何媒体、网站或个人未经阿里云开发者社区协议授权不得转载、链接、转贴或以其他方式复制发布/发表。申请授权请邮件developerteam@list.alibaba-inc.com,已获得阿里云开发者社区协议授权的媒体、网站,在转载使用时必须注明"稿件来源:阿里云开发者社区,原文作者姓名",违者本社区将依法追究责任。 如果您发现本社区中有涉嫌抄袭的内容,欢迎发送邮件至:developer2020@service.aliyun.com 进行举报,并提供相关证据,一经查实,本社区将立刻删除涉嫌侵权内容。

分享:

阿里巴巴开源大数据技术团队成立Apache Spark中国技术社区,定期推送精彩案例,问答区数个Spark技术同学每日在线答疑,只为营造纯粹的Spark氛围,欢迎加入!钉钉群号:21784001,或点击链接申请加入 https://qr.dingtalk.com/action/joingroup?code=v1,k1,jr4ucSRJs6kkK7QZ5sw2oSHRGNoFjONsQgKxZGiTFXA=&_dt_no_comment=1&origin=11

官方博客
官网链接