使用Python開發自訂Processor並部署 - Platform For AI

您可以使用Python開發自訂Processor檔案，結合模型檔案在本地調試後分別上傳至OSS，在部署服務時掛載使用。本文為您介紹使用Python開發自訂Processor的詳細步驟。

背景資訊

說明

建議將模型檔案和Processor的實現分開，這樣在將來需要微調模型並重新部署服務時，可以重用之前的Processor包。您可以在編寫預測邏輯時，使用get_model_path()方法擷取模型的儲存路徑，以便在預測邏輯中載入和使用該模型。
當自訂Processor的依賴較多、Processor包較大時，不建議使用Processor方式來部署服務，建議您使用鏡像方式來部署服務，兩種部署方式的對比介紹請參見部署原理。

使用Python開發自訂Processor的流程如下：

步驟一：構建開發環境
EAS提供的Python SDK支援多種Python機器學習架構，並能夠整合Pandas等資料分析處理架構。通過本文給出的方式，您可以在本地構建Python開發環境，以便後續開發自訂Processor，並打包上傳環境。
步驟二：編寫預測邏輯
EAS的Python SDK提供了高效能的RPC架構和與EAS叢集互動所需的內部介面。您只需要編寫預測邏輯實現簡單的介面，就可以將模型服務部署至EAS。
步驟三：本地測試服務
編寫預測邏輯後，您需要進行本地測試服務。本地測試可以協助您驗證預測邏輯的正確性，以確保後續部署服務後，其功能符合要求。
步驟四：打包代碼檔案與Python環境
按照指定要求打包代碼，以便於後續部署模型服務。
步驟五：上傳Processor程式碼封裝檔案和代碼檔案
將完整的打包檔案和模型檔案分別上傳至OSS。
步驟六：部署和測試服務
將以已構建的自訂的Processor部署為模型服務。

前提條件

已準備好模型檔案。

說明

建議將模型檔案和自訂Processor分開開發，以便於模型和Processor檔案的維護和替換。開發完成後，分別將其上傳到OSS，在後續部署服務時進行掛載調用。

步驟一：構建開發環境

您可以使用Pyenv等Python包管理工具構建開發環境。EAS提供的用戶端工具EASCMD對Python SDK的初始化過程進行了封裝，下載該工具後，您只需要運行一條命令即可完成Python SDK環境初始化，並產生相關的檔案模板，僅適用於Linux作業系統。樣本命令如下：

# 安裝並初始化EASCMD，該樣本為安裝Linux環境的EASCMD工具。
$ wget https://eas-data.oss-cn-shanghai.aliyuncs.com/tools/eascmd/v2/eascmd64
# 下載完成後，可以修改存取權限，配置阿里雲上AccessKey資訊。
$ chmod +x eascmd64
$ ./eascmd64 config -i <access_id> -k <access_key>

# 初始化環境。
$ ./eascmd64 pysdk init ./pysdk_demo

在返回結果中輸入Python版本（預設為3.6版本）後，系統會自動在您配置的./pysdk_demo目錄中建立Python環境ENV目錄、預測服務代碼模板app.py及服務部署模板app.json。

步驟二：編寫預測邏輯

在ENV同級目錄下，建立預測服務主檔案app.py，檔案內容樣本如下：

說明

使用EASCMD構建Python初始化環境時，系統會自動產生app.py檔案，無需手動建立。您可以根據需要修改該檔案內容。
EAS預置的開發鏡像中已預先建立了app.py檔案，無需手動建立，您可以根據需要修改該檔案內容。

# -*- coding: utf-8 -*-
import allspark


class MyProcessor(allspark.BaseProcessor):
    """ MyProcessor is a example
        you can send mesage like this to predict
        curl -v http://127.0.0.1:8080/api/predict/service_name -d '2 105'
    """
    def initialize(self):
        """ load module, executed once at the start of the service
             do service intialization and load models in this function.
        """
        self.module = {'w0': 100, 'w1': 2}
        # model_dir = self.get_model_path().decode()
        # load_model函數需要您自行實現，若您需要載入model.pt模型檔案，則可以實現為torch.load(model_dir + "/model.pt")。
        # self.model = load_model(model_dir)

    def pre_process(self, data):
        """ data format pre process
        """
        x, y = data.split(b' ')
        return int(x), int(y)

    def post_process(self, data):
        """ process after process
        """
        return bytes(data, encoding='utf8')

    def process(self, data):
        """ process the request data
        """
        x, y = self.pre_process(data)
        w0 = self.module['w0']
        w1 = self.module['w1']
        y1 = w1 * x + w0
        if y1 >= y:
            return self.post_process("True"), 200
        else:
            return self.post_process("False"), 400


if __name__ == '__main__':
    # allspark.default_properties().put('rpc.keepalive', '10000')
    # 設定服務計算逾時時間為10s, 預設為5秒
    # parameter worker_threads indicates concurrency of processing
    runner = MyProcessor(worker_threads=10)
    runner.run()

上述代碼即為Python SDK的簡單樣本，代碼中繼承了EAS提供的基類BaseProcessor，實現了initialize()和process()函數。

函數	功能描述	使用說明
initialize()	Processor初始化函數。服務啟動時，進行模型載入等初始化工作。	在`initialize()`函數中添加以下代碼，將模型檔案的擷取與Processor的實現進行分離。 `model_dir = self.get_model_path().decode() self.model = load_model(model_dir)` 通過`get_model_path()`方法擷取bytes類型的模型儲存目錄，即上傳的模型檔案在服務執行個體中的實際儲存目錄。通過自訂`load_model()`函數，來載入和使用模型檔案以實現服務的部署。若您需要載入model.pt模型檔案，則可以實現為`torch.load(model_dir + "/model.pt")`。
get_model_path()	擷取bytes類型的模型儲存目錄。	當您在部署服務的JSON設定檔中通過指定model_path參數來上傳模型檔案時，可調用`get_model_path()`方法擷取模型檔案在服務執行個體中的實際儲存目錄，以用於模型載入。
process(data)	請求處理函數。每個請求會將Request Body作為參數傳遞給process()進行處理，並將函數傳回值返回至用戶端。	data為Request Body，類型為BYTES。傳回值為BYTES和INT類型，輸出參數分別為response_data和status_code，正常請求status_code可以返回0或200。
_init_(worker_threads=5, worker_processes=1,endpoint=None)	Processor建構函式。	worker_threads：Worker線程數，預設值為5。 worker_processes：進程數，預設值為1。如果worker_processes為1，則表示單進程多線程模式。如果worker_processes大於1，則worker_threads只負責讀取資料，請求由多進程並發處理，每個進程均會執行initialize()函數。 endpoint：服務監聽的Endpoint，通過該參數可以指定服務監聽的地址和連接埠，例如endpoint='0.0.0.0:8079'。說明此處配置的連接埠需避開EAS的系統監聽連接埠8080和9090。
run()	啟動服務。	無參數。

步驟三：本地測試服務

在終端app.py檔案所在目錄執行以下命令，啟動Python應用程式。

./ENV/bin/python app.py

系統返回如下結果，表明應用程式啟動成功。

[INFO] waiting for service initialization to complete...
[INFO] service initialization complete
[INFO] create service
[INFO] rpc binds to predefined port 8080
[INFO] install builtin handler call to /api/builtin/call
[INFO] install builtin handler eastool to /api/builtin/eastool
[INFO] install builtin handler monitor to /api/builtin/monitor
[INFO] install builtin handler ping to /api/builtin/ping
[INFO] install builtin handler prop to /api/builtin/prop
[INFO] install builtin handler realtime_metrics to /api/builtin/realtime_metrics
[INFO] install builtin handler tell to /api/builtin/tell
[INFO] install builtin handler term to /api/builtin/term
[INFO] Service start successfully

重新開啟一個終端，執行以下命令測試應用程式的響應情況。
按照步驟二給出的範例程式碼，執行以下命令嚮應用程式發送兩個請求資料，返回結果與代碼邏輯對應。
```
curl http://127.0.0.1:8080/test  -d '10 20'
```

步驟四：打包代碼檔案與Python環境

EASCMD為您提供了打包命令，您可以一鍵完成打包。如果您沒有使用EASCMD進行自訂Processor開發，您也可以手動打包完整環境。兩種打包方式的介紹如下。

使用EASCMD封裝的打包命令（僅Linux系統）。

$ ./eascmd64 pysdk pack ./pysdk_demo

命令執行成功後，輸出如下結果：

[PYSDK] Creating package: /home/xi****.lwp/code/test/pysdk_demo.tar.gz

手動打包

打包要求	詳情
程式碼封裝格式要求	您需要打包為一個.zip或.tar.gz格式的壓縮包。
程式碼封裝內容要求	打包的根目錄必須為/ENV目錄且包含app.py檔案。程式碼封裝樣本：.tar.gz程式碼封裝樣本。

步驟五：上傳Processor程式碼封裝檔案和代碼檔案

打包完成後，您需要將完整的打包檔案（.zip或.tar.gz格式）和模型檔案分別上傳至OSS，便於後續部署服務時掛載使用。上傳至OSS的操作請參見命令列工具ossutil命令參考。

步驟六：部署和測試服務

您可以通過控制台或EASCMD用戶端來部署模型服務。

部署服務。

通過控制台方式部署服務

進入部署服務頁面，詳情請參見服務部署：控制台。

在部署服務頁面配置以下關鍵參數，更多參數配置詳情，請參見服務部署：控制台。

參數	描述
部署方式	選擇模型+processor部署服務。
模型檔案	選擇合適的方式配置模型檔案。
Processor種類	選擇自訂processor。
Processor語言	選擇python。
Processor包	單擊OSS檔案匯入，並選擇上述步驟已打包的檔案所在的OSS路徑。
Processor主檔案	配置為`./app.py`。

（可選）在對應配置編輯中添加data_image參數，參數內容配置為上述步驟中打包時配置的鏡像路徑。
說明
如果您在步驟四中通過鏡像上傳環境，則需要配置data_image參數，否則忽略該操作即可。
單擊部署。

通過EASCMD方式部署服務

以Linux作業系統為例，說明部署服務的操作步驟：

下載EASCMD用戶端並進行身份認證。具體操作，請參見下載並認證用戶端。

在用戶端檔案所在目錄建立JSON格式的檔案，命名為app.json。通過EASCMD或手動打包完整環境的檔案內容樣本如下：

{
  "name": "pysdk_demo",
  "processor_entry": "./app.py",
  "processor_type": "python",
  "processor_path": "oss://examplebucket/exampledirectory/pysdk_demo.tar.gz",
  "model_path": "oss://examplebucket/exampledirectory/model",
  "cloud": {
        "computing": {
            "instance_type": "ecs.c7.large"
        }
  },
  "metadata": {
    "instance": 1,
    }
}

開啟終端工具，在JSON檔案所在目錄，使用以下命令部署服務。

$ ./eascmd64 create app.json

系統返回如下結果，表示服務部署成功。

[RequestId]: 1202D427-8187-4BCB-8D32-D7096E95B5CA
+-------------------+-------------------------------------------------------------------+
| Intranet Endpoint | http://182848887922****.vpc.cn-beijing.pai-eas.aliyuncs.com/api/predict/pysdk_demo |
|             Token | ZTBhZTY3ZjgwMmMyMTQ5OTgyMTQ5YmM0NjdiMmNiNmJkY2M5ODI0****          |
+-------------------+-------------------------------------------------------------------+
[OK] Waiting task server to be ready
[OK] Fetching processor from [oss://eas-model-beijing/195557026392****/pysdk_demo.tar.gz]
[OK] Building image [registry-vpc.cn-beijing.aliyuncs.com/eas/pysdk_demo_cn-beijing:v0.0.1-20190806082810]
[OK] Pushing image [registry-vpc.cn-beijing.aliyuncs.com/eas/pysdk_demo_cn-beijing:v0.0.1-20190806082810]
[OK] Waiting [Total: 1, Pending: 1, Running: 0]
[OK] Service is running

測試服務。
1. 進入EAS頁面，具體操作，請參見服務部署：控制台。
2. 單擊目標服務的服務方式列下的調用資訊，擷取調用的公網地址和Token。
3. 在終端中，根據步驟b查詢的資訊進行調用。
```
$ curl <service_url> -H 'Authorization: <token>' -d '10 20'
```
  其中：
  - <service_url>：替換為步驟b中擷取的公網地址。例如：http://182848887922****.vpc.cn-beijing.pai-eas.aliyuncs.com/api/predict/pysdk_demo。
  - <token>：替換為步驟b中擷取的Token。例如：ZTBhZTY3ZjgwMmMyMTQ5OTgyMTQ5YmM0NjdiMmNiNmJkY2M5ODI0****。
  - -d後的內容配置為服務的調用輸入參數。

Platform For AI：使用Python開發自訂Processor