spark1.6遷移到spark2.2實踐和新特性初探

來自專欄 數據科學家從業之路

1. 前言

如果你沒用過1.6版本的Spark，看這篇文章會很輕鬆，直接關心2.2版本的特性就好。

先簡單介紹下Spark的幾個重要模塊

1. Spark SQL用於處理結構化數據，可以用SQL查詢，可以讀寫HIVE表和HDFS文件，和Hadoop生態無縫連接，這個也是Spark如此流行的原因之一。有評測表示Spark SQL的速度是HiveQL的10倍多，因為Spark SQL在整個數據處理過程中都是在內存中進行，不需要受限於磁碟的IO。快也是有代價的，Spark任務本身是非常耗內存的，也經常會出現內存不足的狀況，好多的任務優化都是圍繞優化內存。
2. MLlib是Spark的機器學習庫。包括常用的分類、回歸、聚類和協同過濾演算法。特徵工程上支持特徵提取、轉換、降維、特徵選擇功能。支持Pipeline，提供模型構建、評估和調優的工具。支持模型保存、下載。另外，該模塊還支持線性代數、統計和數據處理等操作。
3. Spark Streaming用於處理實時數據，輸入可以是Kafka, Flume, Kinesis, 或者TCP sockets，輸出可以是分散式文件系統HDFS、資料庫、面板。中間可以用map、reduce、join和window函數處理，甚至可以用MLlib和GraphX模塊做處理，非常強大了。
4. Spark GraphX用於圖計算，是較新的模塊。谷歌最早的演算法PageRank用于衡量圖中節點的權重，求解過程就是圖計算。

除了Spark Graphx，其他幾個模塊在工業界都是常用的。Spark Streaming筆者用的不多，沒有遷移經驗，主要講Spark SQL、MLlib的變化。

2. Spark SQL模塊

如果你只用Spark處理離線數據，那麼看這個小節就足夠了。

將python版本的Spark1.6的數據流程遷移到Spark2.2版本，一個簡單數據流程的1.6和2.2版本代碼如下：

# spark1.6版本#!/bin/python#coding:utf-8from pyspark import SparkContextfrom pyspark.sql import HiveContextimport sysimport timereload(sys)sys.setdefaultencoding(utf8)sc = SparkContext(appName="yourSpk")hiveContext = HiveContext(sc)df = hiveContext("SELECT * FROM srcTable")sc.stop()#####################版本分隔符######################### spark2.2版本#!/bin/python#coding:utf-8from pyspark import SparkContextfrom pyspark.sql import HiveContextimport sysimport timereload(sys)sys.setdefaultencoding(utf8)spark = SparkSession .builder .appName("yourSpk") .enableHiveSupport() .getOrCreate()df = spark.sql("SELECT * FROM srcTable")spark.stop()

如果你的數據流程非常簡單，且要從spark-1.6遷移到2.2版本，可以參考上面的例子。

2.1 SparkSession作為入口

在Spark1.6中SQLContext 和 HiveContext是處理結構化數據(行列數據)的程序入口，在spark-2.2中用SparkSession代替了上述兩種方式作為程序入口。

其中enableHiveSupport()表示允許操作Hive數據。getOrCreate()表示如果已經存在這個SparkSession則直接返回，否則就按照配置創建一個。

如果需要在創建SparkSession的時候增加配置，用config("spark.some.config.option", "some-value")函數，比如要配置並行度為200，就是config("spark.default.parallelism","200")。當然，這個配置也可以在spark-submit的時候去指定，那就是

spark-submit --conf spark.default.parallelism=200 ${your_sparkfile}.py

效果是一樣的，都是讓spark任務的並行度是200。

2.2 其他變化

DataFrame

• registerTempTable表示根據RDD創建一個臨時表，在2.x中建議用createOrReplaceTempView取代。
• unionAll和SQL中的UNION ALL含義相同，在2.x中建議用union取代。

計算精度

這是個坑。Spark2.2中SELECT 0.1=0 返回的是true! 因為String類型的小數(在這裡是0.1)會被轉換為int，所以被轉換成了0。如果你的數據類型是String類型，做數值計算時會有問題，解決方法是先進行類型轉換。

3. MLlib

Machine Learning Library(MLlib)是Spark的機器學習庫，發生了如下的主要變化。

DataFrame-based APIs的線性代數類

Spark線性代數依賴移到了一個新的工程mllib-local中，線性代數的類移到了spark.ml.linalg，所以DataFrame-based APIs依賴於spark.ml.linalg類。RDD-based API的依賴並沒有改變，還是依賴spark.mllib.linalg。

若想將spark.mllib.linalg的向量和矩陣轉換成spark.ml.linalg形式的向量和矩陣，可以用spark.mllib.util.MLUtils。

Python版本的轉換如下：

from pyspark.mllib.util import MLUtils# convert DataFrame columnsconvertedVecDF = MLUtils.convertVectorColumnsToML(vecDF)convertedMatrixDF = MLUtils.convertMatrixColumnsToML(matrixDF)# convert a single vector or matrixmlVec = mllibVec.asML()mlMat = mllibMat.asML()

Scala版本：

import org.apache.spark.mllib.util.MLUtils// convert DataFrame columnsval convertedVecDF = MLUtils.convertVectorColumnsToML(vecDF)val convertedMatrixDF = MLUtils.convertMatrixColumnsToML(matrixDF)// convert a single vector or matrixval mlVec: org.apache.spark.ml.linalg.Vector = mllibVec.asMLval mlMat: org.apache.spark.ml.linalg.Matrix = mllibMat.asML

Java版本：

import org.apache.spark.mllib.util.MLUtils;import org.apache.spark.sql.Dataset;// convert DataFrame columnsDataset<Row> convertedVecDF = MLUtils.convertVectorColumnsToML(vecDF);Dataset<Row> convertedMatrixDF = MLUtils.convertMatrixColumnsToML(matrixDF);// convert a single vector or matrixorg.apache.spark.ml.linalg.Vector mlVec = mllibVec.asML();org.apache.spark.ml.linalg.Matrix mlMat = mllibMat.asML();

API介面從RDD變成DataFrame

MLLib的API從基於RDD變成基於DataFrame，RDD方式的API以後處於維護階段，會修復bug，但不再增加新功能，RDD API預計在2.3中不推薦使用，在3.0中刪除。官方給出變化的原因是DataFrame更加友好，且在各種編程語言中API形式一致。

新增特性

其他增加的新特性見Spark2.2 MLlib新特性。

比如增加了ALS演算法，可以用於推薦系統的top-k推薦(SPARK-19535)。

被刪除的方法

從1.6到2.0版本中spark.mllib and spark.ml中有不少被刪除或者不推薦使用的方法。在此不再贅述，詳情見1.6到2.0中被刪除或不推薦方法。

(完)