不同的工具

在機器學習的常用工具中，一般的資料探勘和資料統計分析的工具，是R語言和Python，大量的資料時候，使用的是Flink和Spark。
瞭解和熟悉工具的使用，對於一些資料進行探索和實現。
 本文主要是基於Python的資料探勘和機器學習的流程，來對比Spark和Flink的機器學習包，進而通過使用其中的一種情況而熟悉其他，達到觸類旁通的效果

Python

 一般流程： 獲取資料 -> 資料預處理 -> 訓練建模 -> 模型評估 -> 預測，分類
scikit-learn ：  NumPy  SciPy  matplotlib
  管道機制實現了對全部步驟的流式化封裝和管理（streaming workflows with pipelines）
      許多演算法模型串聯起來，比如將特徵提取、歸一化、分類組織在一起形成一個典型的機器學習問題工作流 程式設計技巧的創新，而非演算法的創新
     Transformer 轉換器  Estimator 估計器  Pipeline 管道
  具體
     01.Transformer 轉換器 (StandardScaler，MinMaxScaler)
     02.Estimator 估計器（LinearRegression、LogisticRegression、LASSO、Ridge），
        所有的機器學習演算法模型，都被稱為估計器
     03.Pipeline 管道 將Transformer、Estimator 組合起來成為一個大模型
    	 pipeline
        使用PipeLine對資料進行預處理組成新的模型
        直接呼叫fit和predict方法來對pipeline中的所有演算法模型進行訓練和預測
    	可以結合grid search對引數進行選擇
 示例
     eg： from sklearn.pipeline import Pipeline
     過程：
      資料歸一化(Data Normalization)  from sklearn import preprocessing
      特徵選擇(Feature Selection)     from sklearn.ensemble import ExtraTreesClassifier
      演算法的使用                      from sklearn.linear_model import LogisticRegression
      優化演算法引數                    from sklearn.grid_search import GridSearchCV 
     one-hot編碼
	 資料集拆分
	 模型：
	  # 擬合模型
      model.fit(X_train, y_train)
     # 模型預測
      model.predict(X_test)    
     # 獲得這個模型的引數
      model.get_params()
	 模型儲存和載入
	  from sklearn.externals import joblib
	# 儲存模型
	  joblib.dump(model, 'model.pickle')
	#載入模型
	  model = joblib.load('model.pickle')
 

Spark

1.基本概念

org.apache.spark.ml 
PipelineStage
A stage in a pipeline, either an [[Estimator]] or a [[Transformer]].
Transformer
transform one dataset into another.
Estimator
estimators that fit models to data.
Model
A fitted model, i.e., a [[Transformer]] produced by an [[Estimator]].
Pipeline
A Pipeline consists of a sequence of stages, each of which is either an [[Estimator]] or a [[Transformer]]

PipelineModel
 object PipelineModel extends MLReadable[PipelineModel]
Parameter 
 被用來設定 Transformer 或者 Estimator 的引數
VectorAssembler
   CrossValidatorModel
        Params for [[CrossValidator]] and [[CrossValidatorModel]].
		Spark提供在org.apache.spark.ml.tuning包下提供了模型選擇器，可以替換引數然後比較模型輸出
 

2.Spark 的 Dataset

randomSplit
Randomly splits this Dataset with the provided weights.

 randomSplitAsList
 Returns a Java list that contains randomly split Dataset with the provided weights.
輸入： weights: Array[Double]
       weights: List[Double]
返回： Array[Dataset]or List
示例：
 正樣本和負樣本擷取（樣本資料過多的情況）
                       double[] weights = {pos_rate,1.0-pos_rate};
                       Dataset<Row>[] arr = posSet.randomSplit(weights);
                       posSet = arr[0];
  正樣本和負樣本均衡
//合併正負樣本資料
                   Dataset<Row> dataUse = dataPos_sample.union(dataNeg_sample);   
// 定義 Pipeline 中的各個 PipelineStage ，如指標提取和轉換模型訓練等。
  有了這些處理特定問題的 Transformer 和 Estimator，
 我們就可以按照具體的處理邏輯來有序的組織 PipelineStages 並建立一個 Pipeline
 每個stage要麼是一個Transformer，要麼是一個Estimator。
 這些stage是按照順序執行的，輸入的dataframe當被傳入每個stage的時候會被轉換
 Pipeline pipeline = new Pipeline().setStages(Array(stage1,stage2,stage3,…))
 然後就可以把 訓練資料集 作為入參並呼叫 Pipeline 例項的 fit 方法來開始以流的方式來處理源訓練資料

//構建完成一個 stage piple
    Pipeline pipeline = new Pipeline().setStages(pipeArr);
	PipelineModel model = pipeline.fit(train_data);

    載入模型： PipelineModel model2 = PipelineModel.load(path);
 方式 獲得 CrossValidator 的最佳模型引數 -- 通過交叉驗證進行模型選擇
  CrossValidator rf_cv = new CrossValidator().setEstimator(pipeline)
  CrossValidatorModel rf_model = rf_cv.fit(train_data);
    載入模型： CrossValidatorModel rf_model2 = CrossValidatorModel.load(path);
	  
 eg： // Chain indexers and tree in a Pipeline.
 Pipeline pipeline = new Pipeline()
  .setStages(new PipelineStage[]{labelIndexer, featureIndexer, dt, labelConverter});
 

Flink

1.Flink ML

PipelineStage 
    Base class for a stage in a pipeline，and does not have any actual functionality
    Its subclasses must be either Estimator or Transformer    
Transformer
       * A transformer is a {@link PipelineStage} that transforms an input {@link Table} to a result {@link Table}.   
Estimator
        Estimators are {@link PipelineStage}s responsible for training and generating machine learning models.
Model
       A model is an ordinary {@link Transformer} except how it is created.   
 Pipeline
       A pipeline is a linear workflow which chains {@link Estimator}s and {@link Transformer}s to execute an algorithm.
     can also be used as a {@link PipelineStage} in another pipeline
   
 Params WithParams  ParamInfoFactory  ParamInfo

2.Alink

com.alibaba.alink.pipeline
 Pipeline
     A pipeline is a linear workflow which chains {@link EstimatorBase}s and {@link TransformerBase}s to
  * execute an algorithm.
     public class Pipeline extends EstimatorBase<Pipeline, PipelineModel> 
 PipelineModel
      public class PipelineModel extends ModelBase<PipelineModel> implements LocalPredictable {
 PipelineStageBase
      The base class for a stage in a pipeline, either an [[EstimatorBase]] or a [[TransformerBase]].
 EstimatorBase
    public abstract class EstimatorBase<E extends EstimatorBase<E, M>, M extends ModelBase<M>> extends PipelineStageBase<E> implements Estimator<E, M>
 TransformerBase 
     public abstract class TransformerBase<T extends TransformerBase<T>>  extends PipelineStageBase<T> implements Transformer<T>
 VectorAssembler
     VectorAssembler is a transformer that combines a given list of columns

參考

原始碼