Platform For AI:Python を使用したカスタムプロセッサの開発

背景情報

以下のワークフローは、Python カスタムプロセッサの完全な開発プロセスを説明しています：

1. ステップ 1：開発環境の構築

   Elastic Algorithm Service (EAS) の Python SDK は、複数の機械学習フレームワークをサポートし、Pandas などのデータ処理ライブラリと統合されています。このセクションでは、ローカルの Python 開発環境をセットアップし、必要なプロジェクト構造を作成する方法について説明します。

2. ステップ 2：予測ロジックの作成

   EAS の Python SDK は、高性能な RPC フレームワークとクラスター通信用の内部インターフェイスを提供します。モデルサービスをデプロイするには、リクエストの処理方法を定義する簡単な予測ロジックインターフェイスを実装するだけです。

3. ステップ 3：ローカルテストの実行

   予測ロジックを実装した後、デプロイ前にサービスの正確性を検証し、正常に機能することを確認するためにローカルでテストします。

4. ステップ 4：コードファイルと Python 環境のパッケージング

   EAS のデプロイ要件に従って、コードと Python 環境をパッケージ化します。

5. ステップ 5：パッケージとモデルファイルのアップロード

   パッケージファイルとモデルファイルを個別に OSS にアップロードします。

6. ステップ 6：モデルサービスのデプロイとテスト

   作成したカスタムプロセッサをモデルサービスとしてデプロイします。

前提条件

プロセッサ開発を開始する前に、トレーニング済みのモデルファイルを準備してください。

ステップ 1：開発環境の構築

Pyenv などの Python パッケージマネージャーを使用して、開発環境をセットアップします。EAS が提供する EASCMD クライアントを使用すると、Python SDK の初期化プロセスが簡素化されます。ツールをダウンロードした後、1 つのコマンドを実行するだけで SDK 環境を初期化し、必要なファイルテンプレートを生成できます。このツールは Linux システムでのみ利用可能です。例：

# EASCMD をインストールして初期化します。この例では、Linux 用の EASCMD ツールのインストール方法を示します。
$ wget https://eas-data.oss-cn-shanghai.aliyuncs.com/tools/eascmd/v2/eascmd64
# ダウンロードが完了したら、アクセス権限を変更し、Alibaba Cloud の 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 デプロイ設定を自動的に作成します。

ステップ 2：予測ロジックの作成

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


class MyProcessor(allspark.BaseProcessor):
    """ MyProcessor は一例です
        このようにメッセージを送信して予測できます
        curl -v http://127.0.0.1:8080/api/predict/service_name -d '2 105'
    """
    def initialize(self):
        """ モジュールのロード、サービスの開始時に一度だけ実行されます
             この関数でサービスの初期化とモデルのロードを行います。
        """
        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):
        """ データ形式の前処理
        """
        x, y = data.split(b' ')
        return int(x), int(y)

    def post_process(self, data):
        """ 処理後の後処理
        """
        return bytes(data, encoding='utf8')

    def process(self, 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')
    # サービスのコンピューティングタイムアウトを 10 秒に設定します。デフォルトは 5 秒です。
    # パラメーター worker_threads は処理の同時実行数を示します
    runner = MyProcessor(worker_threads=10)
    runner.run()

model_dir = self.get_model_path().decode()
self.model = load_model(model_dir)

ステップ 3：サービスのローカルテスト

1. ターミナルで `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

2. 新しいターミナルを開き、次のコマンドを実行してサービスをテストします：

   ステップ 2 のサンプルコードに基づいて、2 つのデータ値をサービスに送信します。応答は実装されたロジックと一致します：

   curl http://127.0.0.1:8080/test  -d '10 20'

ステップ 4：コードファイルと Python 環境のパッケージング

EASCMD はワンクリックのパッケージングコマンドを提供します。プロセッサ開発に EASCMD を使用しなかった場合でも、手動でパッケージングできます。以下に両方の方法を説明します。

ステップ 5：プロセッサパッケージとコードファイルのアップロード

パッケージング後、プロセッサパッケージ (.zip または .tar.gz) とモデルファイルの両方を OSS にアップロードします。これらのファイルはサービスデプロイ時にマウントできます。アップロード手順については、「ossutil コマンドリファレンス」をご参照ください。

ステップ 6：サービスのデプロイとテスト

コンソールまたは EASCMD クライアントのいずれかを使用してモデルサービスをデプロイします。

1. サービスのデプロイ

   コンソール経由でのサービスのデプロイ

   1. PAI コンソールにログインします。ページ上部でリージョンを選択します。次に、目的のワークスペースを選択し、[Elastic Algorithm Service (EAS)] をクリックします。

   2. [サービスをデプロイ] をクリックします。[カスタムモデルのデプロイ] セクションで、[カスタムデプロイ] をクリックします。

   3. [カスタムデプロイ] ページで、次の主要なパラメーターを設定します。パラメーター設定の詳細については、「カスタムデプロイ」をご参照ください。

      パラメーター	説明
      [デプロイ方法]	[プロセッサベースのデプロイ] を選択します。
      [モデル設定]	[設定タイプ] を [Object Storage Service (OSS)] に設定し、モデルファイルが配置されている OSS パスを選択します。
      プロセッサータイプ	[カスタムプロセッサ] を選択します。
      プロセッサ言語	[python] を選択します。
      プロセッサ パッケージ	[設定タイプ] を [Object Storage Service (OSS)] に設定し、前のステップでパッケージ化したファイルの OSS パスを選択します。
      [プロセッサのメインファイル]	./app.py に設定します。

   4. (オプション) [サービス設定] セクションで、data_image パラメーターを追加します。パラメーターをステップ 4 のパッケージング時に設定したイメージパスに設定します。

      説明

      ステップ 4 でランタイムイメージを使用して環境をアップロードした場合にのみ、data_image パラメーターを設定してください。

   5. [デプロイ] をクリックします。

   EASCMD を使用したサービスのデプロイ

   以下の手順では、Linux オペレーティングシステムを例として、サービスをデプロイする方法について説明します：

   1. EASCMD クライアントをダウンロードし、身分認証を実行します。詳細については、「クライアントのダウンロードと認証」をご参照ください。

   2. クライアントファイルが配置されているディレクトリで、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,
          }
      }

   3. ターミナルを開きます。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

2. サービスのテスト

   1. PAI コンソールにログインします。ページ上部でリージョンを選択します。次に、目的のワークスペースを選択し、[Elastic Algorithm Service (EAS)] をクリックします。

   2. 対象サービスの [サービスメソッド] 列で、[呼び出し情報] をクリックして、呼び出し用のパブリックエンドポイントと [トークン] を取得します。

   3. ターミナルで、前のステップで取得した情報に基づいて呼び出しを行います。

      $ curl <service_url> -H 'Authorization: <token>' -d '10 20'

      場所:

      • <service_url>：前のステップで取得したパブリックエンドポイントに置き換えます。例： http://182848887922****.vpc.cn-beijing.pai-eas.aliyuncs.com/api/predict/pysdk_demo。

      • <token>：前のステップで取得したトークンに置き換えます。例： ZTBhZTY3ZjgwMmMyMTQ5OTgyMTQ5YmM0NjdiMmNiNmJkY2M5ODI0****。

      • -d の後の内容は、サービス呼び出しの入力パラメーターです。

参考文献

• カスタムプロセッサの開発時に発生する可能性のある問題の詳細については、「よくある質問」をご参照ください。

• Java を使用してカスタムプロセッサを開発する方法の詳細については、「Java を使用したカスタムプロセッサの開発」をご参照ください。