基於機器學習的DAG調度平臺

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

DAG調度平臺的定義及場景

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

系統架構

構圖：

系統架構說明

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

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

1.機器學習前端交互

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

• 數據讀取/數據預處理
• 特徵功能
• 模型訓練
• 模型預測
• 模型評估
• 模型部署

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

2.DAG調度平臺各模塊介紹

dag engine（圖引擎）：

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

opertor engine（算子執行引擎）:

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

executor（執行器）：

• local executor(本地執行器)：

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

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

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

• base executor (執行器接口):

之後的執行器實現須要實現這個基類，方便拓展。

3.分佈式計算平臺交互

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

部署架構圖

separation方式

mixture方式

實現細節

yaml定義格式

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

dag engine實現邏輯

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

operator engine實現邏輯

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

local executor實現邏輯

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