用戶自定義函數(shù)（UDF）是大多數(shù)SQL環(huán)境的一個關(guān)鍵特性，其主要用于擴(kuò)展系統(tǒng)的內(nèi)置功能。UDF允許開發(fā)人員通過抽象其低級語言實(shí)現(xiàn)在更高級語言（如SQL）中應(yīng)用的新函數(shù)。Apache Spark也不例外，其為UDF與Spark SQL工作流集成提供了各種選項。

十多年的張家川回族自治網(wǎng)站建設(shè)經(jīng)驗，針對設(shè)計、前端、開發(fā)、售后、文案、推廣等六對一服務(wù)，響應(yīng)快，48小時及時工作處理。成都營銷網(wǎng)站建設(shè)的優(yōu)勢是能夠根據(jù)用戶設(shè)備顯示端的尺寸不同，自動調(diào)整張家川回族自治建站的顯示方式，使網(wǎng)站能夠適用不同顯示終端，在瀏覽器中調(diào)整網(wǎng)站的寬度，無論在任何一種瀏覽器上瀏覽網(wǎng)站，都能展現(xiàn)優(yōu)雅布局與設(shè)計，從而大程度地提升瀏覽體驗。創(chuàng)新互聯(lián)公司從事“張家川回族自治網(wǎng)站設(shè)計”,“張家川回族自治網(wǎng)站推廣”以來，每個客戶項目都認(rèn)真落實(shí)執(zhí)行。

在本篇博文中，我們將回顧Python、Java和Scala上的Apache Spark UDF和UDAF（用戶自定義的聚合函數(shù)）實(shí)現(xiàn)的簡單示例。我們還將討論重要的UDF API功能和集成點(diǎn)，包括各發(fā)行版本之間的當(dāng)前可用性?？偠灾?，我們將介紹一些重要的性能注意事項，使您對應(yīng)用程序中利用UDF的選擇有所了解。

Spark SQL UDFs

UDF轉(zhuǎn)換了表中單個行的數(shù)值，為每行生成單個對應(yīng)的輸出值。例如，大多數(shù)的SQL環(huán)境都提供了一個UPPER函數(shù)，同時返回了一個大寫版本的字符串以作為輸入。

自定義函數(shù)可以在Spark SQL中定義和注冊為UDF，并具有可用于SQL查詢的關(guān)聯(lián)別名。下面我們將為您介紹一個簡單的例子，我們將定義一個UDF將以下JSON數(shù)據(jù)中的溫度值從攝氏度（Celsius）轉(zhuǎn)換為華氏度（Fahrenheit）：

下面的示例代碼使用SQL別名CTOF注冊我們的轉(zhuǎn)換UDF，然后使用它從SQL查詢中轉(zhuǎn)換每個城市的溫度值。為簡潔起見，省略了SQLContext對象和其他樣板代碼的創(chuàng)建，并在每個代碼段下面提供了完整列表的鏈接。

Python

Scala

Java

請注意，Spark SQL定義了UDF1～UDF22類別，支持包含最多22個輸入?yún)?shù)的UDF。上面的例子中使用UDF1處理單個溫度值作為輸入。如果未能對Apache Spark源代碼進(jìn)行更新，使用數(shù)組（arrays）或結(jié)構(gòu)體（structs）作為參數(shù)對于需要超過22個輸入的應(yīng)用程序可能很有幫助；從風(fēng)格的角度來看，如果您發(fā)現(xiàn)自己使用的是UDF6或更高版本，這一方案可能是選。

Spark SQL UDAF函數(shù)

用戶自定義聚合函數(shù)（UDAF）可以同時處理多行，然后返回單個值作為結(jié)果，其通常與GROUP BY語句（例如COUNT或SUM）共同使用。為了讓示例簡單明了，我們將實(shí)現(xiàn)一個別名為SUMPRODUCT的UDAF按使用分組、給定價格和庫存中的整數(shù)數(shù)量計算所有車輛的零售價值：

目前，Apache Spark UDAF的實(shí)現(xiàn)定義在擴(kuò)展繼承的了UserDefinedAggregateFunction類別中并有由Scala和Java語法支持。一旦定義好之后，我們可以在別名SUMPRODUCT下舉例說明并注冊我們的SumProductAggregateFunction UDAF對象，并從SQL查詢中予以使用，這與前面示例中的CTOF UDF大致相同。

Scala

Apache Spark中的其他UDF支持

Spark SQL支持UDF、UDAF和UDTF等現(xiàn)有Hive（Java或Scala）函數(shù)實(shí)現(xiàn)的集成。順便提醒一下，UDTFs（用戶自定義表函數(shù)）可以返回多個列和行 – 這超出了本文的范圍，但是我們可能在以后的博文中涉及。對于使用前面示例中強(qiáng)調(diào)的方法重新實(shí)現(xiàn)和注冊相同邏輯，集成現(xiàn)有的Hive UDF是有價值的一種替代方法；從性能角度來看，該方法對于PySpark也是有幫助的，這將在下一節(jié)中討論。通過包含Hive UDF函數(shù)實(shí)現(xiàn)的JAR文件利用spark-submit的-jars選項，可以從HiveContext中訪問Hive函數(shù)；然后使用CREATE TEMPORARY FUNCTION對函數(shù)進(jìn)行聲明（如在Hive [1]中所做，包含一個UDF），具體示例如下所述：

Java 中的Hive UDF定義

從Python訪問HiveUDF

請注意，如上文中我們實(shí)現(xiàn)的UDF和UDAF函數(shù)，Hive UDF只能使用Apache Spark的SQL查詢語言進(jìn)行調(diào)用 – 也就是說，不能與Dataframe API的域特定語言（DSL）一起使用。

或者，通過包含實(shí)現(xiàn)jar文件（使用含有spark-submit的-jars選項），以Scala和Java語言實(shí)現(xiàn)的UDF可以從PySpark中進(jìn)行訪問，然后通過SparkContext對象的私有引用執(zhí)行器JVM、底層Scala或裝載在jar文件中的Java UDF實(shí)現(xiàn)來訪問UDF定義。Holden Karau在一次精彩的演講中[2]對這種方法進(jìn)行了探討。請注意，在此技術(shù)中所使用的一些Apache Spark私有變量不是正式面向終端用戶的。這樣做還帶來了額外的好處，允許將UDAF（目前必須在Java和Scala中定義）用于PySpark，下文中的示例中使用了前面在Scala中定義的SUMPRODUCT UDAF進(jìn)行證明：

Scala UDAF定義

Scala UDAF from PySpark

UDF相關(guān)的功能正在連續(xù)不斷地添加至Apache Spark的每一個版本中。例如2.0版本在R中增加了對UDF的支持。作為參考，下表總結(jié)了本文中討論的各版本的關(guān)鍵特性：



表格中匯總了目前為止本博客中介紹過的相關(guān)版本的關(guān)鍵特性。

性能注意事項

了解Apache Spark UDF功能的性能影響是非常重要的。例如，Python UDF（比如我們的CTOF函數(shù)）導(dǎo)致數(shù)據(jù)在運(yùn)行UDF邏輯的執(zhí)行器JVM和Python注釋器之間被序列化 - 與Java或Scala中的UDF實(shí)現(xiàn)相比，這大大降低了性能。緩解這種序列化瓶頸的潛在解決方案包括以下方面：

1. 如上一節(jié)所述，從PySpark中訪問Hive UDF。Java UDF實(shí)現(xiàn)可以由執(zhí)行器JVM直接訪問。請再次注意，這種方法只用于從Apache Spark的SQL查詢語言訪問UDF。
2. 這種方法的使用也可以參考PySpark訪問在Java或Scala中執(zhí)行的UDF，如我們之前定義的Scala UDAF示例所示。

一般來說，UDF邏輯應(yīng)盡可能的精簡，因為可能每一行都會被調(diào)用。例如，在擴(kuò)展到10億行時，UDF邏輯中的一個步驟需要耗費(fèi)100毫秒的時間才能完成，從而很快就會導(dǎo)致重大的要性能問題。

Spark SQL的另一個重要組成部分是Catalyst查詢優(yōu)化器。這一功能隨著每個版本而擴(kuò)展，通?？梢詾镾park SQL查詢提供顯著的性能改進(jìn)；然而，任意UDF實(shí)現(xiàn)代碼對于Catalyst而言可能不是很好理解（雖然分析字節(jié)碼的未來功能[3]被認(rèn)為可以解決這一問題）。因此，使用Apache Spark內(nèi)置SQL查詢函數(shù)功能通?？梢詭碜罴研阅?，并且應(yīng)該是在避免引入UDF時考慮的第一種方法。高級用戶尋求利用Catalyst與其代碼更緊密地結(jié)合，可以參考以下Chris Fregly的演講[4]，該演講人使用Expression.genCode優(yōu)化UDF代碼，并且使用了新的Apache Spark 2.0實(shí)驗功能[5]，其為定制Catalyst優(yōu)化程序規(guī)則提供了一個可即插即用的API。

結(jié)論

當(dāng)Spark SQL的內(nèi)置功能需要擴(kuò)展時，UDF是一個非常有用的工具。本篇博文中提供了一次UDF和UDAF實(shí)現(xiàn)的演練，并討論了其集成步驟，以在Spark SQL中利用Spark SQL中現(xiàn)有的Java Hive UDF。UDF可以在Python、Scala、Java和（在Spark 2.0中）R中實(shí)現(xiàn)，同時UDAF 可以在以及Scala和Java的UDAF中實(shí)現(xiàn)。當(dāng)在PySpark中使用UDF時，必須考慮數(shù)據(jù)序列化成本，并且應(yīng)該考慮采用上文所討論的兩個策略來解決這個問題。最后，我們探討了Spark SQL的Catalyst優(yōu)化器，以及基于性能考慮的因素，在解決方案中引入UDF之前堅持使用內(nèi)置SQL函數(shù)的性能考慮因素。

代碼https://github.com/curtishoward/sparkudfexamples
CDH版本：5.8.0（Apache Spark 1.6.0）

另外有需要云服務(wù)器可以了解下創(chuàng)新互聯(lián)scvps.cn，海內(nèi)外云服務(wù)器15元起步，三天無理由+7*72小時售后在線，公司持有idc許可證，提供“云服務(wù)器、裸金屬服務(wù)器、高防服務(wù)器、香港服務(wù)器、美國服務(wù)器、虛擬主機(jī)、免備案服務(wù)器”等云主機(jī)租用服務(wù)以及企業(yè)上云的綜合解決方案，具有“安全穩(wěn)定、簡單易用、服務(wù)可用性高、性價比高”等特點(diǎn)與優(yōu)勢，專為企業(yè)上云打造定制，能夠滿足用戶豐富、多元化的應(yīng)用場景需求。

分享文章：在ApacheSpark中使用UDF-創(chuàng)新互聯(lián)
標(biāo)題URL：http://jinyejixie.com/article36/dsippg.html

成都網(wǎng)站建設(shè)公司_創(chuàng)新互聯(lián)，為您提供App開發(fā)、建站公司、服務(wù)器托管、網(wǎng)站設(shè)計、營銷型網(wǎng)站建設(shè)、品牌網(wǎng)站設(shè)計

廣告

聲明：本網(wǎng)站發(fā)布的內(nèi)容（圖片、視頻和文字）以用戶投稿、用戶轉(zhuǎn)載內(nèi)容為主，如果涉及侵權(quán)請盡快告知，我們將會在第一時間刪除。文章觀點(diǎn)不代表本網(wǎng)站立場，如需處理請聯(lián)系客服。電話：028-86922220；郵箱：631063699@qq.com。內(nèi)容未經(jīng)允許不得轉(zhuǎn)載，或轉(zhuǎn)載時需注明來源： 創(chuàng)新互聯(lián)