Apache Spark 1.5新特性介紹

apache spark社群剛剛釋出了1.5版本,大家一定想知道這個版本的主要變化,這篇文章告訴你答案。

dataframe執行後端優化(tungsten第一階段)

dataframe可以說是整個spark項目最核心的部分,在1.5這個開發周期内最大的變化就是tungsten項目的第一階段已經完成。主要的變化是由spark自己來管理記憶體而不是使用jvm,這樣可以避免jvm gc帶來的性能損失。記憶體中的java對象被存儲成spark自己的二進制格式,計算直接發生在二進制格式上,省去了序列化和反序列化時間。同時這種格式也更加緊湊,節省記憶體空間,而且能更好的估計資料量大小和記憶體使用情況。如果大家對這部分的代碼感興趣,可以在源代碼裡面搜尋那些unsafe開頭的類即可。在1.4版本隻提供unsafeshufflemanager等少數功能,剩下的大部分都是1.5版本新加入的功能。

其他優化還包括預設使用code generation; cache-aware算法對join,aggregation, shuffle, sorting的增強;window function性能的提高等。

那麼性能到底能提升多少呢?可以參考databricks給出的這個例子。這是一個16 million行的記錄,有1 million的組合鍵的aggregation查詢分别使用spark 1.4和1.5版本的性能對比,在這個測試中都是使用的預設配置。

那麼如果我們想自己測試下tungsten第一階段的性能該如何測試呢?spark 1.4以前的版本中spark.sql.codegen, spark.sql.unsafe.enabled等幾個參數在1.5版本裡面合并成spark.sql.tungsten.enabled并預設為true,隻需要修改這一個參數就可以配置是否開啟tungsten優化(預設是開啟的)。

dataframe/sql/hive

在dataframe api方面,實作了新的聚合函數接口aggregatefunction2以及7個相應的build-in的聚合函數,同時基于新接口實作了相應的udaf接口。新的聚合函數接口把一個聚合函數拆解為三個動作:initialize/update/merge,然後使用者隻需要定義其中的邏輯既可以實作不同的聚合函數功能。spark的這個新的聚合函數實作方法和impala裡面非常類似。

spark内置的expressionfunction得到了很大的增強,實作了100多個這樣的常用函數,例如string, math, unix_timestamp, from_unixtime, to_date等。同時在處理nan值的一些特性也在增強,例如 nan = nan 傳回true;nan大于任何其他值等約定都越來越符合sql界的規則了。

使用者可以在執行join操作的時候,指定把左邊的表或者右邊的表broadcast出去。因為基于cardinality的估計并不是每次都是很準的,如果使用者對資料了解可以直接指定哪個表更小進而被broadcast出去。

hive子產品最大的變化是支援連接配接hive 1.2版本的metastore,同時支援metastore partition pruning(通過spark.sql.hive.metastorepartitionpruning=true開啟,預設為false)。因為很多公司的hive叢集都更新到了1.2以上,那麼這個改進對于需要通路hive中繼資料的spark叢集來說非常重要。spark 1.5支援可以連接配接hive 0.13, 0.14, 1.0/0.14.1, 1.1, 1.2的metastore。

在external data source方面,對parquet的支援有了很大的加強。parquet的版本更新到1.7;更快的metadatadiscovery和schema merging;同時能夠讀取其他工具或者庫生成的非标準合法的parquet檔案;以及更快更魯棒的動态分區插入。

由于parquet更新到1.7,原來的一個重要bug被修複,是以spark sql的filter pushdown預設改為開啟狀态(spark.sql.parquet.filterpushdown=true),能夠幫助查詢過濾掉不必要的io。

spark 1.5可以通過指定spark.sql.parquet.output.committer.class參數選擇不同的output committer類,預設是org.apache.parquet.hadoop.parquetoutputcommitter,使用者可以繼承這個類實作自己的output committer。由于hdfs和s3這兩種檔案存儲系統的差別,如果需要向s3裡面寫入資料,可以使用directparquetoutputcommitter,能夠有效提高寫效率,進而加快job執行速度。

另外還有一些改動,包括:structtype支援排序功能;timestamptype的精度減小到1us;spark現在的checkpoint是基于hdfs的,從1.5版本開始支援基于memory和local disk的checkpoint。這種類型的checkpoint性能更快,雖然不如基于hdfs的可靠,但是對于疊代型機器學習運算還是很有幫助的。

機器學習mllib

mllib最大的變化就是從一個機器學習的library開始轉向建構一個機器學習工作流的系統,這些變化發生在ml包裡面。mllib子產品下現在有兩個包:mllib和ml。ml把整個機器學習的過程抽象成pipeline,一個pipeline是由多個stage組成,每個stage是transformer或者estimator。

以前機器學習工程師要花費大量時間在training model之前的feature的抽取、轉換等準備工作。ml提供了多個transformer,極大提高了這些工作的效率。在1.5版本之後,已經有了25+個featuretransformer,其中countvectorizer, discrete cosine transformation,minmaxscaler, ngram, pca, rformula, stopwordsremover, and vectorslicer這些featuretransformer都是1.5版本新添加的,做機器學習的朋友可以看看哪些滿足你的需求。

這裡面的一個亮點就是rformula的支援,目标是使使用者可以把原來用r寫的機器學習程式(目前隻支援glm算法)不用修改直接搬到spark平台上來執行。不過目前隻支援集中簡單的r公式(包括'.', '~', '+'和'-'),社群在接下來的版本中會增強這項功能。

另外越來越多的算法也作為estimator搬到了ml下面,在1.5版本中新搬過來的有naive bayes, k-means, isotonic regression等。大家不要以為隻是簡單的在ml下面提供一個調用相應算法的api,這裡面變換還是挺多的。例如naive bayes原來的模型分别用array[double]和array[array[double]]來存儲pi和theta,而在ml下面新的api裡面使用的是vector和matrix來存儲。從這也可以看出,新的ml架構下所有的資料源都是基于dataframe,所有的模型也盡量都基于spark的資料類型表示。在ml裡面的publicapi下基本上看不到對rdd的直接操作了,這也與tungsten項目的設計目标是一緻的。

除了這些既有的算法在ml api下的實作,ml裡面也增加了幾個新算法:

● multilayerperceptronclassifier(mlpc)這是一個基于前饋神經網絡的分類器,它是一種在輸入層與輸出層之間含有一層或多層隐含結點的具有正向傳播機制的神經網絡模型,中間的節點使用sigmoid (logistic)函數,輸出層的節點使用softmax函數。輸出層的節點的數目表示分類器有幾類。mlpc學習過程中使用bp算法,優化問題抽象成logisticloss function并使用l-bfgs進行優化。

● mllib包裡面增加了一個頻繁項挖掘算法prefixspan,associationrules能夠把freqitemset生成關聯式規則。

● 在mllib的統計包裡面實作了kolmogorov–smirnov檢驗,用以檢驗兩個經驗分布是否不同或一個經驗分布與另一個理想分布是否不同。

另外還有一些現有算法的增強:lda算法,決策樹和ensemble算法,gmm算法。

● ml裡面的多個分類模型現在都支援預測結果的機率而不像過去隻支援預測結果,像logisticregressionmodel, naivebayesmodel,decisiontreeclassificationmodel, randomforestclassificationmodel,gbtclassificationmodel等,分别使用predictraw,predictprobability, predict分别可以得到原始預測、機率預測和最後的分類預測。同時這些分類模型也支援通過設定thresholds指定各個類的門檻值。

● randomforestclassificationmodel和randomforestregressionmodel模型都支援輸出feature importance

● gmm em算法實作了當feature次元或者cluster數目比較大的時候的分布式矩陣求逆計算。實驗表明當feature次元>30,cluster數目>10的時候,這個優化性能提升明顯。

● 對于linearregressionmodel和logisticregressionmodel實作了linearregressiontrainingsummary和logisticregressiontrainingsummary用來記錄模型訓練過程中的一些統計名額。

1.5版本的python api也在不斷加強,越來越多的算法和功能的python api基本上與scala api對等了。此外在tuning和evaluator上也有增強。

其他

從1.5開始,standalone, yarn和mesos三種部署方式全部支援了動态資源配置設定。

sparkr支援運作在yarn叢集上,同時dataframe的函數也提供了一些r風格的别名,可以降低熟悉r的使用者的遷移成本。

在streaming和graphx方面也有非常大的改進,在這裡不在一一贅述,詳細可以參考release note。

原文釋出時間為：2015-09-16

本文來自雲栖社群合作夥伴“大資料文摘”，了解相關資訊可以關注“bigdatadigest”微信公衆号