什么是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去執行對應算子。

總結

以上是生活随笔為你收集整理的python dag调度系统开发_基于机器学习的DAG调度平台的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。