摘要：算子背后是實現的一些算法組件機器學習前端交互機器學習平臺前端主要是將機器學習的流程裝成一個，定義各個算子的出入參，以及算子的配置參數，組裝成一個文件，傳給調圖平臺是方式交互，是通過文件定義，通過。

什么是DAG?
有向無環圖
樹形結構：除根節點，每個節點有且僅有一個上級節點，下級節點不限。根節點沒有上級節點。
圖結構：每個節點上級、下級節點數不限。

任務調度是在各行各業是個基礎問題，當任務復雜同時存在任務復雜依賴的時候，就需要DAG調度。如：機器學習的可視化建模（PAI平臺、第四范式），數據的抽取、轉換、加載（ETL），在業務復雜情況下就需要DAG的調度管理等
接下來說說基于機器學習的DAG調度平臺

構圖：

DAG調度平臺主要的職責是:
1.接受機器學習web傳過來的yaml文件(dag定義文件)
2.解析yaml文件，變成結構化數據存儲到mysql數據庫
3.開始調度dag定義各個算子任務
4.算子執行引擎根據算子類型分發到各個環境進行執行

名詞說明
yaml:類型XML的數據描述語言，語法更加簡單
算子:機器學習的DAG中各個節點即為算子，在算子執行引擎中稱為算子任務。算子背后是python實現的一些算法組件

機器學習平臺前端主要是將機器學習的流程裝成一個dag，定義各個算子的出入參，以及算子的配置參數，組裝成一個yaml文件，傳給DAG調圖平臺（Azkaban是zip方式交互，Airflow是通過py文件定義，Oozie通過xml）。
一個完整的DAG定義應包含以下算子：

數據讀取/數據預處理

特征功能

模型訓練

模型預測

模型評估

模型部署

下圖是個簡化版的DAG定義，除去了模型部署算子

dag engine（圖引擎）：

負責解析傳入的yaml文件。根據yaml的配置生成算子的出入參以及運行配置信息保存到數據庫。同時負責任務的調用。

opertor engine（算子執行引擎）:

負責算子執行，根據算子類型分發到不同的執行器中。統一的啟停接口，日志查詢接口，任務狀態查詢接口

executor（執行器）：

local executor(本地執行器)：

執行單機的python任務，執行單機文件方式的機器學習算法。當沒有大數據平臺的時候，只能通過本地執行器執行DAG

dc executor（分布式計算平臺執行器）：

將python算法發送至大數據計算平臺，使用大數據平臺資源運行算子。

base executor (執行器接口):

以后的執行器實現需要實現這個基類，方便拓展。

針對不同的的計算平臺實現base executor去自定義擴充。本系統通過dc executor實現，
分布式計算平臺需要將python code通過http接口發送過去進行執行。

dag:
 operator_list: [algo_local_read_file_45_1517360824080,algo_local_split_data_45_1517360836712,algo_local_model_2c_l_45_1517362008544,algo_local_model_predict_45_1517362016532,algo_local_model_2c_eval_45_1517362022452,algo_local_model_gbdt_111_1517801573063]
 operator_rels:
  algo_local_read_file_45_1517360824080: [{"target":"algo_local_split_data_45_1517360836712","source_index":0,"target_index":0}]
  algo_local_split_data_45_1517360836712: [{"target":"algo_local_model_2c_l_45_1517362008544","source_index":0,"target_index":0},{"target":"algo_local_model_gbdt_111_1517801573063","source_index":1,"target_index":0}]
  algo_local_model_predict_45_1517362016532: [{"target":"algo_local_model_2c_eval_45_1517362022452","source_index":0,"target_index":0}]
  algo_local_model_gbdt_111_1517801573063: [{"target":"algo_local_model_predict_45_1517362016532","source_index":0,"target_index":0}]
  algo_local_model_2c_l_45_1517362008544: [{"target":"algo_local_model_predict_45_1517362016532","source_index":0,"target_index":1}]
 operator_details:
  
  algo_local_read_file_45_1517360824080:
   algo_name: algo_local_read_file
   data_type: 本地python
   type: 數據源
   cn_name: 讀文件
   coordinate:
    x: 137
    y: 69
   params:
    data_id: 40
  algo_local_split_data_45_1517360836712:
   algo_name: algo_local_split_data
   data_type: 本地python
   type: 數據預處理
   cn_name: 拆分組件
   coordinate:
    x: 226
    y: 164
   params:
    split_type: 1
    ext1: 0.8
    ext2: null
  algo_local_model_2c_l_45_1517362008544:
   algo_name: algo_local_model_2c_l
   data_type: 本地python
   type: 模型算法
   cn_name: 邏輯回歸二分類
   coordinate:
    x: 130
    y: 262
   params:
    x_cols: [LIMIT_BAL,SEX,EDUCATION,MARRIAGE,AGE,PAY_0,PAY_2,PAY_3,PAY_4,PAY_5,PAY_6,BILL_AMT1,BILL_AMT2,BILL_AMT3,BILL_AMT4,BILL_AMT5,BILL_AMT6,PAY_AMT1,PAY_AMT2,PAY_AMT3,PAY_AMT4,PAY_AMT5,PAY_AMT6]
    y_col: next_month
    pre_value: 1
    penalty: l2
    C: 1
    max_iter: 100
    senior: true
    class_weight: null
    dual: false
    fit_intercept: true
    intercept_scaling: 1
    multi_class: ovr
    n_jobs: 1
    random_state: null
    solver: liblinear
    tol: 0.0001
    verbose: 0
    warm_start: false
  algo_local_model_predict_45_1517362016532:
   algo_name: algo_local_model_predict
   data_type: 本地python
   type: 模型預測
   cn_name: 模型預測
   coordinate:
    x: 258
    y: 396
   params:
    x_cols: [LIMIT_BAL,SEX,EDUCATION,MARRIAGE,AGE,PAY_0,PAY_2,PAY_3,PAY_4,PAY_5,PAY_6,BILL_AMT1,BILL_AMT2,BILL_AMT3,BILL_AMT4,BILL_AMT5,BILL_AMT6,PAY_AMT1,PAY_AMT2,PAY_AMT3,PAY_AMT4,PAY_AMT5,PAY_AMT6]
  algo_local_model_2c_eval_45_1517362022452:
   algo_name: algo_local_model_2c_eval
   data_type: 本地python
   type: 模型評估
   cn_name: 二分類評估
   coordinate:
    x: 270
    y: 503
   params:
    y_col: next_month
    pre_col: predict_result
    pre_value: 1
  algo_local_model_gbdt_111_1517801573063:
   algo_name: algo_local_model_gbdt
   data_type: 本地python
   type: 模型算法
   cn_name: GBDT
   coordinate:
    x: 432.1111111111111
    y: 295.3333333333333
   params:
    x_cols: [LIMIT_BAL,SEX,EDUCATION,MARRIAGE,AGE,PAY_0,PAY_2,PAY_3,PAY_4,PAY_5,PAY_6,BILL_AMT1,BILL_AMT2,BILL_AMT3,BILL_AMT4,BILL_AMT5,BILL_AMT6,PAY_AMT1,PAY_AMT2,PAY_AMT3,PAY_AMT4,PAY_AMT5,PAY_AMT6]
    y_col: next_month
    pre_value: 1
    n_estimators: 10
    max_depth: 5
    senior: true
    criterion: friedman_mse
    init: null
    learning_rate: 0.1
    loss: deviance
    max_features: null
    max_leaf_nodes: null
    min_impurity_decrease: 0
    min_impurity_split: null
    min_samples_leaf: 1
    min_samples_split: 2
    min_weight_fraction_leaf: 0
    presort: auto
    random_state: null
    subsample: 1
    verbose: 0
    warm_start: false
 params:
  translate: [41,-20]
  scale: 0.9

1.當前節點，采用廣度優先遍歷獲取所有需要執行的算子(節點)信息。
2.輪詢所有算子(節點)，判斷上算子(節點)是否全部執行完成，執行完成開始執行當前算子(節點)。
3.發送請求到operator engine開始執行當前算子(節點)任務。

1.主進程接受task請求，添加任務執行隊列、任務監聽隊列。
2.任務執行進程輪詢接受到的隊列，根據不同任務類型調用不同executor
3.任務監聽進程輪詢接受到的隊列，調用不同executor查詢任務執行狀態，是任務執行的最終狀態（成功、失敗）回調dag engine

1.local executor接受任務，發送到隊列中。
2.local worker進程池(cpu數*2個進程)，輪詢獲取隊列中任務，使用importlib的python去執行對應算子。