Create a task that contains a custom operator

Step 1. Upload the custom operator

Uploading a custom operator involves three steps:

• Import the development framework provided by the platform into your operator code.

• Package your operator into an image and upload it to the Alibaba Cloud ACR repository.

• Upload the image in the Custom Operator Management module.

The following sections describe these steps in detail.

Step 1.1. Import the platform's development framework

1. Install the SDK

• Create an `sdk` folder in the project's root directory. Then, download the SDK file and save it to the `sdk` folder:

  ali_autodrive-0.0.1.tar.gz

• Install the module: `pip install sdk/ali_autodrive-0.0.1.tar.gz`

2. Implement the abstract class

• Create a module, such as `example`, in the project's root directory. In the `example` module, create a data processing class, such as `DataProcessor`.

• `DataProcessor` must implement the `DataProcessTaskAbstract` abstract class and override the partition and processing methods.

import json
import time

import av
import os.path

from abc import ABC
from ali_autodrive.parallel_compute.model.FileContent import FileContent
from ali_autodrive.parallel_compute.utils.tree_util import *
from ali_autodrive.parallel_compute.DataProcessTaskAbstract import DataProcessTaskAbstract

MAX_RECORD_NUM = 200


class DataProcessor(DataProcessTaskAbstract, ABC):
    def __init__(self):
        super(DataProcessor, self).__init__()

    def data_partition(self, context):
        self.get_logger().info("Data processor, data partition start.")
        file_tree_reader = FileTreeReader(self.get_file_tree())
        while file_tree_reader.finish is False:
            file_list = file_tree_reader.get_sub_node_file_list(1)
            for file in file_list:
                if "file_path" in file and file["file_path"].endswith('.mp4'):
                    self.save_data_partition([json.dumps(file)])

        self.get_logger().info("Data processor, data partition end.")

    def data_process(self, context):
        self.get_logger().info("Data processor, data process start.")

        # Extract video frames
        local_file_list = get_sub_node_local_file_list(self.get_file_tree())
        for file_path in local_file_list:
            self.framing(file_path)

        self.get_logger().info("Data processor, data process end.")

    def framing(self, file_path):
        file_name = self.__get_file_name(file_path)
        container = av.open(file_path)
        stream = container.streams.video[0]
        stream.codec_context.skip_frame = 'NONREF'
        path = self.get_user_workspace() + "/img/" + file_name
        if not os.path.exists(path):
            os.makedirs(path)

        file_list = []
        timestamp_ns = time.time_ns()
        for frame in container.decode(stream):
            file_path = path + "/frame-%04d.jpg" % frame.index
            frame.to_image().save(file_path)

            # Tag the file
            file_tag = {}
            file_tag["timestamp_ns"] = timestamp_ns + int(frame.time * 1000000000)
            content = FileContent()
            content.file_tag = file_tag
            content.file_path = file_path

            if len(file_list) >= MAX_RECORD_NUM:
                self.save_data_partition(file_list)
                file_list = []
            else:
                file_list.append(json.dumps(content.__dict__))
        if len(file_list) > 0:
            self.save_data_partition(file_list)

    @staticmethod
    def __get_file_name(file_path):
        return file_path.split("/")[-1].split(".")[0]

Step 1.2. Perform a local test

1. Create a configuration file

Create a configuration file, such as `config.ini`, in the project's root directory.

[init]
# The working directory. Replace this with the workspace on your machine.
workspace = /Users/icyore/workspace
# The simulated OSS address. Replace this with your test file directory.
ossInput = /Users/icyore/oss/input 
# The simulated OSS address. Replace this with your test output directory.
ossOutput = /Users/icyore/oss/output
# [Optional] The operator initialization parameter. You can get this parameter in the operator. It is the same as the "Transform Program Parameter" of the standardized transform node in the data management platform. The parameter is a JSON string.
transformParams ={"vehicleId":"parallel computing test vehicle"}

2. Create a startup script

Create a startup script, such as `test_start.py`, in the project's root directory.

from ali_autodrive.parallel_compute.service_startup import *

# ./config.ini: The path of the configuration file. It can be a relative or absolute path.
# example.DataProcessor: The module where the operator is located.
# DataProcessor: The operator implementation class.
test("./config.ini", "example.DataProcessor", "DataProcessor")

3. Run the startup script

Run the startup script.

python test_start.py

Step 1.3. Create an image

1. Package the module

Create a `setup.py` file in the project's root directory. Run the packaging command: python setup.py sdist

The packaged file is saved to the `dist` folder.

# -*- coding:utf-8 -*-
from setuptools import (setup, find_packages)

setup(
    # Package name
    name="example",
    # Version
    version="0.0.1",
    # List of sub-packages to include
    packages=find_packages(),
    # Add dependencies
    install_requires=[
        #'python-lzf==0.2.4',
    ]
)

2. Create a Dockerfile

Create a Dockerfile in the project's root directory.

FROM python:3.8
COPY . /app
WORKDIR /app

RUN pip install sdk/ali_autodrive-0.0.1.tar.gz
ADD   sdk/ali_autodrive-0.0.1.tar.gz  ali_autodrive
RUN pip install dist/example-0.0.1.tar.gz
WORKDIR /app/ali_autodrive/ali_autodrive-0.0.1/ali_autodrive/parallel_compute

EXPOSE 5000
# example.DataProcessor: The module where the operator is located.
# DataProcessor: The operator implementation class.
CMD ["python","service_startup.py" ,"example.DataProcessor","DataProcessor"]

3. Upload the image

Build the image and upload it to ACR.

docker login --username=jier****@city-brain-pro auto-driver-registry.cn-hangzhou.cr.aliyuncs.com
docker build -t parallel-compute-example:0.0.1 .
docker tag parallel-compute-example:0.0.1 auto-driver-registry.cn-hangzhou.cr.aliyuncs.com/partition_compute/parallel-compute-example:0.0.1
docker push auto-driver-registry.cn-hangzhou.cr.aliyuncs.com/partition_compute/parallel-compute-example:0.0.1

`auto-driver-registry.cn-hangzhou.cr.aliyuncs.com` is the ACR repository address and `parallel_compute` is the ACR namespace. Replace them with your repository address and namespace. You must also log on to your account.

`parallel-compute-example:0.0.1` is the image name and version number. You can customize the name and version number.

Step 1.4. Upload the custom operator

• In the navigation pane on the left, find Custom Operator Management in the Data Definition module.

• In the upper-right corner, click Add Operator. Select a category and follow the prompts to upload the operator.

• In the operator list, find the operator that you uploaded and click Publish.

Step 2. Create a task that contains the custom operator

After you complete the preceding steps, the operator that you uploaded is available in the System Nodes list on the Task Configuration page. You can use the custom operator by itself or in an orchestration with built-in operators.

For detailed instructions, see Create a simple data processing task using a built-in operator.