Spark MLlib 學習入門筆記

阿新 • • 發佈：2019-01-15

RDD（Resilient Distributed Datasets）分散式彈性資料集，將資料分佈儲存在不同節點的計算機記憶體中進行儲存和處理。RDD的任務被分成兩部分：Transformation和Action。Transformation用於對RDD的建立，即一個RDD轉換為另一個RDD，Action是資料計算執行部分，如count、reduce、collect等。 Spark文件裡有相關的說明，網上還有一個Spark文件的中文翻譯，可以參考。從程式設計的角度來看，一開始把RDD當成一個數組，並記住它的執行任務由Transformation運算元和Action運算元共同完成，得到運算結果就好了。

以下是我在學習RDD時完成的一道練習題，通過這道練習，可以掌握RDD程式設計的基本思路和方法。

對機器學習資料iris.data資料集進行簡單的處理，包括filter、count、distince和分類統計（求和、最大值、最小值和平均值）。

1. 資料說明

[plain] view plain copy

5.1,3.5,1.4,0.2,Iris-setosa
4.9,3.0,1.4,0.2,Iris-setosa
4.7,3.2,1.3,0.2,Iris-setosa
4.6,3.1,1.5,0.2,Iris-setosa
5.0,3.6,1.4,0.2,Iris-setosa

第一個資料septal length；第二個資料sepal width；第三個資料petal length；第四個資料petal width；第五個資料class標籤；在程式中用name表示，相當於key。

2.原始碼

Iris.scala

[plain] view plain copy

package basic.iris
/**
* Created by Oliver on 2017/5/18.
*/
//1. sepal length in cm
//2. sepal width in cm
//3. petal length in cm
//4. petal width in cm
//5. class:
class Iris extends java.io.Serializable {
var sl: Double = 0
var sw: Double = 0
var pl: Double = 0
var pw: Double = 0
var sl_max: Double = 0
var sw_max: Double = 0
var pl_max: Double = 0
var pw_max: Double = 0
var sl_min: Double = 0
var sw_min: Double = 0
var pl_min: Double = 0
var pw_min: Double = 0
var name: String =""
var count: Int = 1
override def toString : String = {
var s = "(" + name + ", "
s = s + "[ " + "%.1f".format(sl) + ", " + "%.1f".format(sl_min) + ", " + "%.1f".format(sl_max) + ", " + "%.1f".format(sl/count) + " ], "
s = s + "[ " + "%.1f".format(sw) + ", " + "%.1f".format(sw_min) + ", " + "%.1f".format(sw_max) + ", " + "%.1f".format(sw/count) + " ], "
s = s + "[ " + "%.1f".format(pl) + ", " + "%.1f".format(pl_min) + ", " + "%.1f".format(pl_max) + ", " + "%.1f".format(pl/count) + " ], "
s = s + "[ " + "%.1f".format(pw) + ", " + "%.1f".format(pw_min) + ", " + "%.1f".format(pw_max) + ", " + "%.1f".format(pw/count) + " ], "
s = s + count.toString + ")"
return s
}
}

IrisStat.scala

[plain] view plain copy

package basic.iris
import org.apache.spark.{SparkConf, SparkContext}
/**
* Created by Oliver on 2017/5/18.
*/
//local
//E:/MyProject/SparkDiscover/data/iris.data
object IrisStat {
def isValid(line: String): Boolean = {
val parts = line.split(",")
return parts.length == 5
}
def parseLine(line: String): (String, Iris) = {
val parts = line.split(",")
val iris = new Iris
iris.sl = parts(0).toDouble
iris.sw = parts(1).toDouble
iris.pl = parts(2).toDouble
iris.pw = parts(3).toDouble
iris.sl_min = parts(0).toDouble
iris.sw_min = parts(1).toDouble
iris.pl_min = parts(2).toDouble
iris.pw_min = parts(3).toDouble
iris.sl_max = parts(0).toDouble
iris.sw_max = parts(1).toDouble
iris.pl_max = parts(2).toDouble
iris.pw_max = parts(3).toDouble
iris.name = parts(4)
return (iris.name, iris)
}
def add(a: Iris, b: Iris): Iris = {
val c = new Iris
c.sl = a.sl + b.sl
c.sw = a.sw + b.sw
c.pl = a.pl + b.pl
c.pw = a.pw + b.pw
c.count = a.count + b.count
c.name = a.name
//比較大小
c.sl_max = math.max(a.sl_max, b.sl_max)
c.sw_max = math.max(a.sw_max, b.sw_max)
c.pl_max = math.max(a.pl_max, b.pl_max)
c.pw_max = math.max(a.pw_max, b.pw_max)
c.sl_min = math.min(a.sl_min, b.sl_min)
c.sw_min = math.min(a.sw_min, b.sw_min)
c.pl_min = math.min(a.pl_min, b.pl_min)
c.pw_min = math.min(a.pw_min, b.pw_min)
return c
}
def printResult(res: (String, Iris)){
println(res._2)
}
def main(args: Array[String]){
val conf = new SparkConf().setMaster(args(0)).setAppName("Iris")
val sc = new SparkContext(conf)
val data = sc.textFile(args(1)).filter(isValid(_))
// distinct
println("---1---------------------------")
data.map(_.split(",")(4)).distinct().foreach(println)
// 簡單計數
val c = data.count()
val c_setosa = data.filter( "Iris-setosa" == _.split(",")(4) ).count()
val c_versicolor = data.filter( "Iris-versicolor" == _.split(",")(4) ).count()
val c_virginica = data.filter( "Iris-virginica" == _.split(",")(4) ).count()
println("")
println("---2---------------------------")
println(c, c_setosa, c_versicolor, c_virginica)
// mapreduce 分組求和、求平均、求最大最小
//data.map(parseLine(_)).foreach(println)
println("")
println("---3---------------------------")
data.map(parseLine(_)).reduceByKey(add(_,_)).collect().foreach(printResult)
}
}

3.執行配置

0?wx_fmt=png

4. 執行結果

[plain] view plain copy

---1---------------------------
Iris-setosa
Iris-versicolor
Iris-virginica
---2---------------------------
(150,50,50,50)
---3---------------------------
(Iris-setosa, [ 250.3, 4.3, 5.8, 5.0 ], [ 170.9, 2.3, 4.4, 3.4 ], [ 73.2, 1.0, 1.9, 1.5 ], [ 12.2, 0.1, 0.6, 0.2 ], 50)
(Iris-versicolor, [ 296.8, 4.9, 7.0, 5.9 ], [ 138.5, 2.0, 3.4, 2.8 ], [ 213.0, 3.0, 5.1, 4.3 ], [ 66.3, 1.0, 1.8, 1.3 ], 50)
(Iris-virginica, [ 329.4, 4.9, 7.9, 6.6 ], [ 148.7, 2.2, 3.8, 3.0 ], [ 277.6, 4.5, 6.9, 5.6 ], [ 101.3, 1.4, 2.5, 2.0 ], 50)

5.程式碼說明

1)Iris.scala

Iris類存放資料檔案的5個欄位，當count為1是單個物件的資料，即資料檔案的一行。當count大於1時，表示對多行資料的對應欄位進行求和運算後的結果，count行計數，_min和_max欄位存放的是對應欄位的最小值和最大值。過載了toString輸出結果。

2)IrisStat.scala

isValid用於判斷資料是否是有效資料，無效則拋棄，完成一個RDD轉換，相當於資料清洗。

parseLine解析資料，將字串轉換為物件，方便後續處理。

add執行運算，包括求和、計數、最大值和最小值，平均值在輸出是用“和”除以“計數”就得到了。

main函式說明：

(1) data = sc.textFile(args(1)).filter(isValid(_)) 過濾無效資料，如果不過濾，只要資料集中有一行錯誤資料，程式就會出錯，如混入一個空行。

(2) distinct ，map是用split得到第5個欄位，然後對第5個欄位應用distince就可以了。

(3) filter，輸入過濾條件就可以了，然後在呼叫count計數。

(4)分組求和、最小值和最大值，調研map傳人parseLine函式轉換得到一個新資料集，這個資料集是(String, Iris)的<key, value>形式，再呼叫reduceByKey就實現了分組操作；在reduceByKey中傳入add執行所定義的執行，呼叫collect返回資料集，用foreach列印結果。

(5)注意下劃線，理解為要傳入的資料就好了，如filter(isValid(_))中，filter把上一個map的資料通過_傳給isValid。

0?wx_fmt=gif

Spark MLlib 學習入門筆記

Spark MLlib 學習入門筆記

Apache Spark MLlib學習筆記（六）MLlib決策樹類演算法原始碼解析 2

Spring MVC學習入門筆記

python3.6。爬蟲學習入門筆記

mangoDb 學習入門筆記

深度學習入門筆記（二）————線性神經網路解決異或問題（程式碼）

深度學習入門筆記系列 ( 二 )——基於 tensorflow 的一些深度學習基礎知識

深度學習入門筆記系列（三）——感知器模型和 tensorboard 的使用方法

機器學習入門筆記一

《Spark機器學習》筆記——基於MovieLens資料集使用Spark進行電影資料分析

程式設計師的機器學習入門筆記（七）：推薦系統入門介紹

Django學習入門筆記(三)

深度學習——入門筆記（3）

深度學習——入門筆記（1）

《Spark機器學習》筆記——Spark分類模型（線性迴歸、樸素貝葉斯、決策樹、支援向量機）

Udacity機器學習入門筆記——自選演算法隨機森林

.net Elasticsearch 學習入門筆記

深度學習入門筆記

500多頁的機器學習入門筆記，下載超5萬次，都講了些什麼？

深度增強學習入門筆記（一）

Spark MLlib 學習入門筆記

相關推薦