Develop and deploy a custom processor using Python - Platform For AI

Workflow

Develop a custom Python processor by completing the following steps:

Step 1: Set up the development environment

The EAS Python SDK supports various Python machine learning frameworks and integrates with data processing frameworks such as Pandas. Set up a local Python environment, develop custom processors, then package and upload the environment.
Step 2: Write the prediction logic

The EAS Python SDK provides a high-performance RPC framework and internal interfaces for interacting with the EAS cluster. Write the prediction logic and implement a simple interface to deploy your model service to EAS.
Step 3: Test the service locally

Test the service locally to verify that the prediction logic is correct and that the service functions as expected after deployment.
Step 4: Package the code and Python environment

Package the code and environment for service deployment.
Step 5: Upload the processor package and model file

Upload the package and model files separately to Object Storage Service (OSS).
Step 6: Deploy and test the service

Deploy your custom processor as a model service and verify that it works.

Note

Separate the model file from the processor to reuse the processor package when you fine-tune and redeploy the model. In the prediction logic, use the get_model_path() method to get the model storage path so the logic can load the model.
If your custom processor has many dependencies or a large package, use an image to deploy the service instead. For a comparison of the two deployment methods, see How it works (Image deployment).

Prerequisites

Prepare a model file.

Note

Develop the model file and custom processor separately. Upload them to OSS separately so they can be mounted during service deployment.

Step 1: Set up development environment

Use Python package management tools such as Pyenv to set up a development environment. EAS provides the EASCMD client tool, which wraps the Python SDK initialization process. After downloading the tool, run a single command to initialize the Python SDK environment and generate file templates. EASCMD supports Linux only. Example:

# Install and initialize EASCMD. This example shows how to install the EASCMD client for Linux.
$ wget https://eas-data.oss-cn-shanghai.aliyuncs.com/tools/eascmd/v2/eascmd64
# After the download is complete, modify the access permissions and configure your Alibaba Cloud AccessKey pair.
$ chmod +x eascmd64
$ ./eascmd64 config -i <access_id> -k <access_key>

# Initialize the environment.
$ ./eascmd64 pysdk init ./pysdk_demo

When prompted, enter a Python version (default: 3.6). The system then creates a Python environment directory named ENV, a prediction service code template named app.py, and a service deployment template named app.json in the ./pysdk_demo directory.

Step 2: Write the prediction logic

In the same directory as ENV, create the main file for the prediction service, named app.py. Example:

Note

If you use EASCMD to initialize the Python environment, app.py is generated automatically. Modify the file content as needed.

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


class MyProcessor(allspark.BaseProcessor):
    """ MyProcessor is an 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 initialization and load models in this function.
        """
        self.module = {'w0': 100, 'w1': 2}
        # model_dir = self.get_model_path().decode()
        # You must implement the load_model function yourself. For example, to load a model.pt file, you can implement it as 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')
    # Set the service timeout to 10s. The default is 5s.
    # The worker_threads parameter indicates the processing concurrency.
    runner = MyProcessor(worker_threads=10)
    runner.run()

This example inherits the BaseProcessor base class provided by EAS and implements the initialize() and process() functions.

Method	Description	Usage
initialize()	Processor initialization method. Performs initialization tasks such as loading the model when the service starts.	Add the following code to `initialize()` to separate model file retrieval from the processor implementation. `model_dir = self.get_model_path().decode() self.model = load_model(model_dir)` `get_model_path()` returns the storage directory of the model file as a bytes object. This is the actual storage directory of the uploaded model file in the service instance. Implement a `load_model()` function to load the model file. For example, to load a model.pt file, implement the function as `torch.load(model_dir + "/model.pt")`.
get_model_path()	Returns the storage directory of the model file as a bytes object.	When the `model_path` parameter is specified in the deployment JSON, call `get_model_path()` to retrieve the model storage directory within the service instance for loading.
process(data)	Request handling method. For each request, the request body is passed to process(). The return value is sent to the client.	data: Request body, type BYTES. Return values are BYTES and INT, corresponding to response_data and status_code. For a normal request, status_code can be 0 or 200.
_init_(worker_threads=5, worker_processes=1,endpoint=None)	Processor constructor.	worker_threads: Number of worker threads. Default value: 5. worker_processes: Number of processes. Default value: 1. If set to 1, the service runs in single-process, multi-thread mode. If greater than 1, worker_threads only read data, and multiple processes handle requests concurrently. Each process executes initialize(). endpoint: Address and port the service listens on. Example: endpoint='0.0.0.0:8079'. Note Do not use port 8080 or 9090, which are reserved by EAS.
run()	Starts the service.	None.

Step 3: Test the service locally

In the directory that contains app.py, run the following command to start the application.

./ENV/bin/python app.py

Expected output on success:

[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

Open a new terminal and run the following command to test the application.

Based on the example code in Step 2, send request data to the application. The returned result reflects the code logic.
```
curl http://127.0.0.1:8080/test  -d '10 20'
```

Step 4: Package code and environment

EASCMD provides a one-step packaging command. If you are not using EASCMD, package the environment manually. Both methods are described below.

EASCMD packaging command (Linux only):

$ ./eascmd64 pysdk pack ./pysdk_demo

Expected output on success:

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

Manual packaging

Requirement	Details
Package format	Package must be a compressed archive in .zip or .tar.gz format.
Package contents	Root directory must be /ENV and must contain app.py. Package example: .tar.gz package example.

Step 5: Upload package and model files

Upload the package file (.zip or .tar.gz) and the model file to OSS separately for mounting during service deployment. For more information, see Quick start for ossutil.

Step 6: Deploy and test the service

Deploy the model service by using the PAI console or EASCMD.

Deploy the service.

PAI console

Log on to the PAI console. Select a region on the top of the page. Then, select the desired workspace and click Elastic Algorithm Service (EAS).
Click Deploy Service. In the Custom Model Deployment section, click Custom Deployment.

On the Custom Deployment page, configure the following parameters. For other parameters, see Custom deployment.

Parameter	Description
Deployment Method	Select Processor-based Deployment.
Model Settings	For Type, select OSS and select the OSS path of the model file.
Processor Type	Select Custom Processor.
Processor Language	Select python.
Processor Package	For Type, select OSS and select the OSS path of the package file from the previous step.
Processor Main File	Enter `./app.py`.

(Optional) In the Service Configurations section, add the data_image parameter and set it to the image path configured during packaging.

Note
This parameter is required only if you used an image to package the environment in Step 4.
Click Deploy.

EASCMD

Deploy the service on Linux:

Download and authenticate EASCMD. For more information, see Download and authenticate the client.

In the EASCMD directory, create a JSON file named app.json. Example:

{
  "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,
    }
}

In the directory containing app.json, run the following command to deploy the service.

$ ./eascmd64 create app.json

Expected output on success:

[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

Test the service.
1. Log on to the PAI console. Select a region on the top of the page. Then, select the desired workspace and click Elastic Algorithm Service (EAS).
2. Find the target service and click Invocation Information in the Service Type column to obtain the public endpoint and Token.
3. Run the following command to call the service with the endpoint and token from the previous step.
```
$ curl <service_url> -H 'Authorization: <token>' -d '10 20'
```
  Parameters:
  - <service_url>: Public endpoint from the previous step. Example: http://182848887922****.vpc.cn-beijing.pai-eas.aliyuncs.com/api/predict/pysdk_demo.
  - <token>: Token from the previous step. Example: ZTBhZTY3ZjgwMmMyMTQ5OTgyMTQ5YmM0NjdiMmNiNmJkY2M5ODI0****.
  - -d: Input data for the service.

References

Common development issues: FAQ
Custom processor development in Java: Develop a custom processor by using Java