秦子實

摘要：隨著機器學習算法在工業(yè)領(lǐng)域的大規(guī)模應用，企業(yè)內(nèi)部網(wǎng)絡(luò)急需部署一個可以用于機器學習算法驗證、調(diào)試以及應用的系統(tǒng)。該系統(tǒng)應具備足夠的算力，同時支持研發(fā)人員并發(fā)運行多個計算任務(wù)，并在計算任務(wù)結(jié)束后返回結(jié)果。該文將使用消息隊列和分布式進程調(diào)度框架設(shè)計一個滿足企業(yè)內(nèi)部需求的分布式機器學習平臺。

關(guān)鍵詞：分布式系統(tǒng)；機器學習；消息隊列

中圖分類號：TP393 文獻標識碼：A 文章編號：1009-3044（2018）01-0201-02

1 概述

隨著人工智能應用的日益成熟，越來越多的工業(yè)領(lǐng)域使用了機器學習算法解決實際問題。在企業(yè)的日常運作中會產(chǎn)生大量的業(yè)務(wù)數(shù)據(jù)，研究人員使用機器學習算法從這些業(yè)務(wù)數(shù)據(jù)中挖掘出有更價值的信息。因此，企業(yè)內(nèi)部中需要一個能夠支持研發(fā)人員并發(fā)運行機器學習算法的平臺，本文將使用Python作為主要編程語言，利用Celery分布式進程調(diào)度系統(tǒng)分配計算任務(wù)、利用Redis臨時存儲計算作業(yè)臨時消息及計算結(jié)果、利用TensorFlow/Keras執(zhí)行機器學習算法，最終通過Flask將平臺發(fā)布為網(wǎng)頁應用，向研發(fā)人員提供機器學習算法驗證、調(diào)試等服務(wù)。

2 Celery技術(shù)簡介

Celery是一個架構(gòu)簡潔、配置靈活且具有高可用性的分布式任務(wù)隊列框架，它擅長通過分布式系統(tǒng)并行處理大量作業(yè)，并提供維護此類系統(tǒng)的必要工具。Celery專注于實時處理任務(wù)隊列，同時也支持任務(wù)調(diào)度。

Celery的架構(gòu)由三部分組成，消息隊列系統(tǒng)（message broker）、任務(wù)執(zhí)行單元（celery worker）和作業(yè)結(jié)果儲存（task store backend）。Celery本身雖然不提供消息服務(wù)，但是兼容大部分常見的消息隊列（如Redis、RabbitMQ等）。Celery worker可以運行在分布式系統(tǒng)的節(jié)點上。作業(yè)結(jié)果可以以內(nèi)存數(shù)據(jù)庫（如Redis、memcached）、數(shù)據(jù)庫（MongoDB）、對象關(guān)系映射（SQLAlchemy、DjangoORM）等多種方式進行數(shù)據(jù)持久化。

此外，Celery還可以與gevent等框架集成，以支持大規(guī)模并發(fā)特性。同時也支持諸如pickle、json、yaml等多種序列化格式。

3 系統(tǒng)結(jié)構(gòu)設(shè)計

分布式機器學習系統(tǒng)通過Celery進行作業(yè)調(diào)度，Celery接受前端的任務(wù)（例如通過Flask接受用戶提交的任務(wù)）后，在多臺運行Celery進程的主機上分發(fā)作業(yè)并執(zhí)行。由于機器學習算法普遍具有運算量大、作業(yè)時間長等特點，所以各執(zhí)行中的作業(yè)應定期將作業(yè)進度回寫至消息隊列或持久化系統(tǒng)中，以在前端（如用戶在Flask上的提交作業(yè)的頁面）實時更新作業(yè)進度。在作業(yè)結(jié)束時Celery worker將結(jié)果寫入消息隊列或持久化系統(tǒng)，前端可以從中讀取結(jié)果。

3 系統(tǒng)結(jié)構(gòu)設(shè)計

本文以Flask項目為例組織代碼，使用Redis作為消息隊列，使用SQLAlchemy進行數(shù)據(jù)持久化，使用TensorFlow/Keras作為機器學習計算平臺。

3.1 Flask項目結(jié)構(gòu)

Flask項目的源碼結(jié)構(gòu)如下：

[- config.py

- run.py

- app/

|- app/__init__.py

|- app/models.py

|- app/views.py

- preprocess/

|- preprocess/__init__.py

|- preprocess/models.py

|- preprocess/views.py

- analysis/

|- analysis/__init__.py

|- analysis/models.py

|- analysis/views.py

|- analysis/datamodel.py

|- analysis/tasks.py ]

一個典型的分布式機器學習項目至少應包含項目配置（config.py）、自動化腳本（run.py）、用戶及權(quán)限（app模塊）、數(shù)據(jù)預處理（preprocess模塊）、模型訓練（analysis模塊）這五部分：

1） 項目配置包括Flask、SQLAlchemy、Redis、Celery、TensorFlow/Keras等模塊的全局配置，以及項目相關(guān)常量（如模型保存路徑、上傳類型限制等）；

2） 自動化腳本包括項目初始化、部署、啟動、停止、升級、調(diào)試等命令行腳本；

3） 用戶及權(quán)限是B/S系統(tǒng)的必備功能，包括用戶-角色的多對多映射、用戶-角色的增刪改等功能；

4） 數(shù)據(jù)預處理模塊負責對用戶上傳的數(shù)據(jù)進行結(jié)構(gòu)化處理，并將結(jié)構(gòu)化數(shù)據(jù)寫入內(nèi)存數(shù)據(jù)庫等持久化系統(tǒng)，以供之后的機器學習算法或前端可視化模塊快速調(diào)用；

5） 模型訓練模塊應具備模型建立（模型結(jié)構(gòu)及模型規(guī)模參數(shù)可配置）、訓練配置（批訓練集及迭代次數(shù)等參數(shù)可調(diào)整）、訓練作業(yè)監(jiān)控（模型誤差及迭代次數(shù)實時消息回寫）、訓練結(jié)果保存（模型及模型運算結(jié)果保存）等機器學習系統(tǒng)常見功能。

3.2 Redis消息隊列

Redis系統(tǒng)部署在消息服務(wù)器上，作為消息中間人（Message Broker）的角色，配合Celery在分布式系統(tǒng)中調(diào)度并發(fā)的機器學習作業(yè)。

此外，Redis本身具備鍵值對內(nèi)存數(shù)據(jù)庫的功能，利用Redis極高的并發(fā)讀寫速度，可以用于暫存Celery作業(yè)的中間狀態(tài)，以供前端實時監(jiān)控機器學習計算作業(yè)的進度。endprint

Redis以消息中間人的形式集成在Celery中：

[# - app/

# |- app/__init__.py

from celery import Celery

celery = Celery（'mltasks'， broker='redis：//：） ]

使用Redis在Flask項目中進行數(shù)據(jù)保存與讀?。?/p>

[from redis import Redis

rds = Redis（host=''）

...

rds.set（result_uuid， result）

...

result = rds.get（result_uuid） ]

3.3 Celery進程調(diào)度

Celery進程運行在分布式系統(tǒng)的所有主機上，各Celery進程通過Redis消息隊列交換信息，協(xié)調(diào)資源。

在Flask項目中，Celery通常在程序入口處初始化：

[# - app/

# |- app/__init__.py

from celery import Celery

celery = Celery（'mltasks'， broker='redis：//：） ]

項目中涉及機器學習模型訓練的代碼應該放在Celery作業(yè)中運行，這部分代碼應集中管理在作業(yè)模塊中，將每一個耗時計算封裝為獨立的函數(shù)，并給函數(shù)添加@celery.task修飾符，供Flask項目代碼異步調(diào)用。

函數(shù)首先使用訓練集訓練模型，訓練過程產(chǎn)生的日志可以通過“l(fā)ogging_uuid”參數(shù)實時寫入Redis消息隊列供前端顯示訓練進度；然后將訓練完成的模型保存至文件系統(tǒng)，同時使用訓練好的模型對驗證集進行預測，以查看模型泛化性能；最后將驗證結(jié)果保存在Redis消息系統(tǒng)中，供前端顯示訓練結(jié)果：

[# - analysis/

# |- analysis/tasks.py

from app import celery

from app import rds

@celery.task

def training_task（data， training_uuid， logging_uuid， epochs）：

model = SomeModel（）

vs = data['validate_set']

ts = data['training_set']

# 使用訓練集訓練模型，并設(shè)置此次訓練的日志回寫地址為logging_uuid

training_history = model.train（ts， logging_uuid， epochs）

# 使用驗證集測試模型

predict = model.predict（vs）

# 保存模型

model.save（）

# 將訓練結(jié)果保存至Redis系統(tǒng)

training_result = json.dumps（{'status'： 'success'，

'predict'： predict}）

rds.set（training_uuid， training_result）

# 將訓練日志狀態(tài)設(shè)為訓練完成，以通知前端更新顯示

logging_result = json.dumps（{'model_state'： 'trained'}）

rds.set（logging_uuid， logging_result） ]

類似的耗時操作，如訓練模型、保存模型、加載模型、使用模型預測等耗時操作均可封裝在上述Celery作業(yè)中，當Flask項目需要執(zhí)行這些耗時操作時，使用Celery作業(yè)提供的“delay”異步調(diào)用：

[# - analysis/

# |- analysis/views.py

from app import celery

from app import rds

from flask import request

from flask import Response

from config import HOST_ID

from analysis.tasks import training_task

import uuid

@app.route（'/api/v1/analysis/somedata//models'， methods=['GET']）

def data_model_training_service（data_id）：

if request.args.get（'action'） == 'train'：

data = some_get_data_function（data_id）

# 為celery的各作業(yè)生成唯一的uuid

training_uuid = HOST_ID + str（uuid.uuid1（））

logging_uuid = HOST_ID + str（uuid.uuid1（））

# 使用celery異步執(zhí)行作業(yè)

training_task.delay（data， training_uuid， logging_uuid， epochs=int（request.args.get（'epochs'）））

# 執(zhí)行其他操作并返回

return Response（json.dumps （{'status'： 'success'}）） ]

上述函數(shù)以異步的方式執(zhí)行模型訓練，執(zhí)行后即刻返回，并指定Redis系統(tǒng)中的日志回寫地址以及訓練結(jié)果回寫地址。使用“celery.task.delay”方式調(diào)用的函數(shù)均運行在Celery worker中，“delay”函數(shù)的參數(shù)列表就是“@celery.task”修飾符修飾對象的參數(shù)列表，每次“delay”調(diào)用均使用獨立進程。

這種異步調(diào)用可以保證在Flask項目中該路由請求不會被耗時操作阻塞，導致前端界面無響應。為了配合使用這類耗時操作的異步調(diào)用，前端及Flask項目需要改變異步操作相關(guān)代碼的編寫模式：

1） 每一個耗時操作的代碼應至少分為兩部分，第一部分負責設(shè)置回寫地址并異步執(zhí)行耗時操作；第二部分負責從日志回寫地址中取回耗時操作執(zhí)行進度。

2） 當前端需要執(zhí)行耗時操作時，向Flask項目發(fā)送開始執(zhí)行的請求，F(xiàn)lask生成唯一的Redis回寫地址并使用“delay”函數(shù)異步調(diào)用，在響應中通知前端回寫地址。

3） 前端程序應定期向Flask項目發(fā)送請求，使用日志回寫地址查詢?nèi)罩拘畔?，并在界面中跟新耗時操作的執(zhí)行進度。

使用日志回寫地址取回耗時操作執(zhí)行進度信息的代碼示例如下：

[# - analysis/

# |- analysis/views.py

from app import rds

@app.route（'/api/v1/analysis/somedata//logs'， methods=['GET']）

def data_model_log_service（data_id）：

return Response（r.get（request.args.get（'redisLoggingTaskID'））） ]

完成各耗時操作函數(shù)的編寫后，將Flask項目及其運行環(huán)境分發(fā)在分布式系統(tǒng)的其他主機上，之后在這些主機上分別啟動celery worker，這些進程將通過Redis主機上的消息隊列交換數(shù)據(jù)并進行進程調(diào)度：

[celery worker -A app.celery --loglevel=debug ]

4 結(jié)束語

本文介紹了使用Celery分布式進程調(diào)度系統(tǒng)搭建用于企業(yè)內(nèi)部網(wǎng)絡(luò)的機器學習計算平臺。該平臺架構(gòu)簡潔，具備較強的可擴展性，容易通過添加主機的方式線性提升系統(tǒng)計算能力，在企業(yè)內(nèi)部的機器學習算法驗證、調(diào)試及應用中發(fā)揮了重要作用，是一種易于部署實現(xiàn)的平臺。