# Alink漫談(十) ：線性迴歸實現 之 資料預處理 [TOC] ## 0x00 摘要 Alink 是阿里巴巴基於實時計算引擎 Flink 研發的新一代機器學習演算法平臺，是業界首個同時支援批式演算法、流式演算法的機器學習平臺。本文和下文將介紹線性迴歸在Alink中是如何實現的，希望可以作為大家看線性迴歸程式碼的Roadmap。 因為Alink的公開資料太少，所以以下均為自行揣測，肯定會有疏漏錯誤，希望大家指出，我會隨時更新。 本系列目前已有十篇，歡迎大家指點 ## 0x01 概念 ### 1.1 線性迴歸 線性迴歸是利用數理統計中迴歸分析，來確定兩種或兩種以上變數間相互依賴的定量關係的一種統計分析方法，運用十分廣泛。其表達形式為y = w'x+e，e為誤差服從均值為0的正態分佈。 線上性迴歸中，目標值與特徵之間存在著線性相關的關係。即假設這個方程是一個線性方程，一個多元一次方程。 基本形式：給定由 d 個屬性描述的示例 ，線性模型試圖學得一個通過屬性的線性組合來進行預測的函式，即： $$ f(x)=w_1x_1 +w_2x_2 ... +w_dx_d+b $$ 其中w為引數，也稱為權重，可以理解為x1，x2...和 xd 對f(x)的影響度。 一般形式為： $$ f(x)=w^Tx+b $$ 假如我們依據這個公式來預測 f(x)，公式中的x是我們已知的，然而w，b的取值卻不知道，只要我們把w，b的取值求解出來，模型就得以確定。我們就可以依據這個公式來做預測了。 那麼如何依據訓練資料求解 w 和 b 的最優取值呢？關鍵是衡量 f 和 y 之間的差別。這就牽扯到另外一個概念：**損失函式（Loss Function）。** ### 1.2 優化模型 假如有一個模型 f(x)，如何判斷這個模型是否優秀？這種定性的判斷可以通過一個成為經驗誤差風險的數值來進行衡量，也就是模型 f 在所有訓練樣本上所犯錯誤的總和 E(x)。 我們通過在訓練集上最小化經驗損失來訓練模型。換言之，通過調節 f 的引數 w，使得經驗誤差風險 E(x) 不斷下降，最終達到最小值的時候，我們就獲得了一個 “最優” 的模型。 但是如果按照上面的定義，E(x) 是一組示性函式的和，因此是不連續不可導的函式，不易優化。為了解決這個問題，人們提出了**“損失函式”**的概念。損失函式就是和誤差函式有一定關係（比如是誤差函式的上界），但是具有更好的數學性質（比如連續，可導，凸性等），比較容易進行優化。所以我們就可以對損失函式來優化。 損失函式如果連續可導，所以我們可以用梯度下降法等一階演算法，也可以用牛頓法，擬牛頓法等二階演算法。當優化演算法收斂後，我們就得到一個不錯的模型。如果損失函式是一個凸函式，我們就可以得到最優模型。 典型的優化方法： | | 一階演算法 | 二階演算法 | | ---------- | ------------------------------------------------------------ | ---------------- | | 確定性演算法 | 梯度下降法 投影次梯度下降 近端梯度下降 Frank-Wolfe演算法 Nesterov加速演算法 座標下降法 對偶座標上升法 | 牛頓法，擬牛頓法 | | 隨機演算法 | 隨機梯度下降法 隨機座標下降法 隨機對偶座標上升法 隨機方差減小梯度法 | 隨機擬牛頓法 | 所以我們可以知道，優化LinearRegression模型 f 的手段一定是：確定損失函式，用 x，y 作為輸入訓練以求得損失函式最小值，從而確定 f 的引數 w。過程大致如下： 1. 處理輸入，把 x, y 轉換成演算法需要的格式。 2. 找一個合適的預測函式，一般表示為 **h** 函式，該函式就是我們需要找的分類函式，它用來預測輸入資料的判斷結果。 3. 構造一個Cost函式（損失函式），該函式表示預測的輸出（**h**）與訓練資料類別（**y**）之間的偏差，可以是二者之間的差（**h-y**）或者是其他的形式。綜合考慮所有訓練資料的 “損失”，將Cost求和或者求平均，記為**J(θ)**函式，表示所有訓練資料預測值與實際類別的偏差。 4. 顯然，損失函式 **J(θ)** 函式的值越小表示預測函式越準確（即**h**函式越準確），所以這一步需要做的是找到 **J(θ)** 函式的最小值。注意，損失函式是關於 **θ** 的函式！也就是說，對於損失函式來講，**θ**不再是函式的引數，而是損失函式的自變數！ 5. 準備模型元資料，建立模型。 ### 1.3 損失函式&目標函式 先概括說明： - 損失函式：計算的是一個樣本的誤差； - 代價函式：是整個訓練集上所有樣本誤差的平均，經常和損失函式混用； - 目標函式：代價函式 + 正則化項； 再詳細闡釋： 假設我們用 f(X) 來擬合真實值Y。這個輸出的f(X)與真實值Y可能是相同的，也可能是不同的，為了表示我們擬合的好壞，我們就用一個函式來度量擬合的程度。這個函式就稱為損失函式(loss function)，或者叫代價函式(cost function)。 損失函式用來衡量演算法的執行情況，估量模型的預測值與真實值的不一致程度，是一個非負實值函式，通常使用 L(Y,f(x)) 來表示。損失函式越小，模型的魯棒性就越好。損失函式是**經驗風險函式**的核心部分。 目標函式是一個相關但更廣的概念，對於目標函式來說在有約束條件下的最小化就是損失函式（loss function）。 因為f(x)可能會過度學習歷史資料，導致它在真正預測時效果會很不好，這種情況稱為過擬合(over-fitting)。這樣得到的函式會過於複雜。所以我們不僅要讓經驗風險最小化，還要讓**結構風險最小化**。這個時候就定義了一個函式 J(x)，這個函式專門用來度量**模型的複雜度**，在機器學習中也叫正則化(regularization)。常用的有 L1, L2範數。 L1 正則的本質是為模型增加了“**模型引數服從零均值拉普拉斯分佈**”這一先驗知識。 L2 正則的本質是為模型增加了“**模型引數服從零均值正態分佈**”這一先驗知識。 L1 正則化增加了所有權重 w 引數的絕對值之和逼迫更多 w 為零，也就是變稀疏（ L2 因為其導數也趨 0, 奔向零的速度不如 L1 給力了）。L1 正則化的引入就是為了完成特徵自動選擇的光榮使命，它會學習地去掉無用的特徵，也就是把這些特徵對應的權重置為 0。 L2 正則化中增加所有權重 w 引數的平方之和，逼迫所有 w 儘可能趨向零但不為零（L2 的導數趨於零）。因為在未加入 L2 正則化發生過擬合時，擬合函式需要顧忌每一個點，最終形成的擬合函式波動很大，在某些很小的區間裡，函式值的變化很劇烈，也就是某些 w 值非常大。為此，L2 正則化的加入就懲罰了權重變大的趨勢。 到這一步我們就可以說我們最終的優化函式是：min(L(Y, f(x) + J(x)) ，即最優化經驗風險和結構風險，而這個函式就被稱為**目標函式**。 在迴歸問題中，通過目標函式來求解最優解，常用的是平方誤差（最小二乘線性迴歸）代價函式。損失函式則是平方損失函式。 ### 1.4 最小二乘法 均方誤差是迴歸任務中最常用的效能度量，因此可以使均方誤差最小。基於均方誤差最小化來進行模型求解的方法稱為“最小二乘法”。線上性迴歸中，最小二乘法就是找到一條直線，使所有樣本到直線的 "歐式距離和" 最小。於是線性迴歸中損失函式就是平方損失函式。 有了這些基礎概念，下面我們就開始動手分析Alink的程式碼。 ## 0x02 示例程式碼 首先，我們給出線性迴歸的示例。 ```java public class LinearRegressionExample { static Row[] vecrows = new Row[] { Row.of("$3$0:1.0 1:7.0 2:9.0", "1.0 7.0 9.0", 1.0, 7.0, 9.0, 16.8), Row.of("$3$0:1.0 1:3.0 2:3.0", "1.0 3.0 3.0", 1.0, 3.0, 3.0, 6.7), Row.of("$3$0:1.0 1:2.0 2:4.0", "1.0 2.0 4.0", 1.0, 2.0, 4.0, 6.9), Row.of("$3$0:1.0 1:3.0 2:4.0", "1.0 3.0 4.0", 1.0, 3.0, 4.0, 8.0) }; static String[] veccolNames = new String[] {"svec", "vec", "f0", "f1", "f2", "label"}; static BatchOperator vecdata = new MemSourceBatchOp(Arrays.asList(vecrows), veccolNames); static StreamOperator svecdata = new MemSourceStreamOp(Arrays.asList(vecrows), veccolNames); public static void main(String[] args) throws Exception { String[] xVars = new String[] {"f0", "f1", "f2"}; String yVar = "label"; String vec = "vec"; String svec = "svec"; LinearRegression linear = new LinearRegression() .setLabelCol(yVar) // 這裡把變數都設定好了，後續會用到 .setFeatureCols(xVars) .setPredictionCol("linpred"); Pipeline pl = new Pipeline().add(linear); PipelineModel model = pl.fit(vecdata); BatchOperator result = model.transform(vecdata).select( new String[] {"label", "linpred"});