Spark UDF的性能的特點是什么

本篇內容主要講解“Spark UDF的性能的特點是什么”，感興趣的朋友不妨來看看。本文介紹的方法操作簡單快捷，實用性強。下面就讓小編來帶大家學習“Spark 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 geohashdef 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負責跑一個個的任務。

性能基準測試

我們創建了一個隨機的經緯度數據集， 包含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 Columnfrom pyspark.sql.column import _to_java_columnfrom pyspark.sql.column import _to_seqfrom 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

重要結論

下面是測試結果

測試結果中， 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

到此，相信大家對“Spark UDF的性能的特點是什么”有了更深的了解，不妨來實際操作一番吧！這里是億速云網站，更多相關內容可以進入相關頻道進行查詢，關注我們，繼續學習！