PAIFlow Manifest参数说明

公共格式

API版本和元信息

apiVersion: 工作流格式的版本
metadata:
  name: 工作流名
  identifier: 工作流标识
  version: 工作流版本
  provider: 工作流提供方
  guid: 唯一ID，用于标识临时节点。
  displayName: 工作流显示名
  annotations: 工作流注解，目前只支持官方注解

 pai.aliyun.com/algo-type:
 alink:
 type: "batch"

输入输出定义

spec:
  inputs: 输入定义
    artifacts: 
    parameters:
  outputs: 输出定义
    artifacts:
    parameters:

• Artifact

  Artifact是数据流，代表机器学习在某一阶段的成果，用来串联上下流的工作流。

  artifacts:
  - name: 名字
    metadata: 类型
    value: 值
    desc: 描述
    repeated: 是否可扩展为多个Artifact
    required: 值是否必填，默认为false

  字段名称	类型	默认值	含义
  name	String	无	名称，在同一个节点内唯一。
  metadata	Object	无	元信息，包含Artifact类型、存储类型等。
  value	String	无	值。
  desc	String	无	描述，可为空。
  repeated	Bool	false	是否可扩展为多个Artifact，这些Artifact metadata相同。 true：可扩展 false：不可扩展
  required	Bool	false	值是否必填。 true：必填 false（默认值）：可选

• Parameter

  Parameter是工作流运行参数。

  parameters:
  - name: 参数名
    type: 参数类型
    value: 参数默认值
    desc: 参数描述
    feasible: 参数规范

  字段名称	类型	默认值	含义
  name	String	无	名称，在同一个节点内唯一。
  type	String	无	支持的参数类型： String Double Int Bool Map List
  value	String	无	参数的默认值，如果没有设置就使用该值。
  desc	String	无	描述
  feasible	Feasible	无	参数的规范，目前支持： min：最小值 max：最大值 len：长度 regex：正则表达式

单节点工作流构造示例

整体定义

工作流只包含一个节点定义，适用于算法定义场景。定义出的单节点工作流可以在DAG工作流中直接引用。

apiVersion: 工作流格式的版本
metadata:
  name: 工作流名
  identifier: 工作流标识
  version: 工作流版本
  provider: 工作流提供方
  displayName: 工作流显示名
  annotations: 工作流注解，目前只支持官方注解
spec:
  inputs: 输入定义
    artifacts: 
    parameters:
  outputs: 输出定义
    artifacts:
    parameters:
  initContainers: 初始化容器列表
  - name: 容器名
    image: 镜像
    command: 运行命令
    args: 运行参数
  container: 主容器描述
    image: 镜像
    command: 运行命令
    args: 运行参数
  sideCarContainers: 容器列表
  - name: 容器名
    image: 镜像
    command: 运行命令
    args: 运行参数  

详细定义

API版本、元信息以及输入输出定义等详情，请参见公共格式。

spec:
  initContainers: 初始化容器列表
  - name: 容器名
    image: 镜像
    command: 运行命令
    args: 运行参数
  container: 主容器
    image: 镜像
    command: 运行命令
    args: 运行参数
  sideCarContainers: 容器列表
  - name: 容器名
    image: 镜像
    command: 运行命令
    args: 运行参数  

示例

以下是输出MaxCompute表信息的单节点工作流定义。

• 有一个输出Artifact，类型是MaxCompute表。

• 有一个输入参数，代表输出MaxCompute表名。

apiVersion: core/v1
metadata:
  provider: '13266******76250'
  version: v1
  identifier: echo
spec:
  inputs:
    parameters:
    - name: outputTableName
      type: String
      value: pai_temp_outputTable
  outputs:
    artifacts:
    - name: outputTable
      metadata:
        type:
          DataSet:
            locationType: MaxComputeTable
      desc: SQL Script Output Port
  container:
    image: 'registry.cn-shanghai.aliyuncs.com/paiflow-core/max-compute-executor:v1.1.4'
    command:
    - sh
    - -c
    args:
    - |
      mkdir -p /pai/outputs/artifacts/outputTable/
      echo '{\"metadata\":{\"type\":{\"DataSet\":{\"locationType\":\"MaxComputeTable\"}}}' > /pai/outputs/artifacts/outputTable/metadata
      echo '{\"location\": {\"endpoint\": \"http://service.cn-shanghai.maxcompute.aliyun.com/api\",\n      \"project\": \"wyl_t******t2\", \"table\":  \"{{inputs.parameters.outputTableName}}\"}}' > /pai/outputs/artifacts/outputTable/value

DAG工作流构造示例

整体定义

工作流定义了一个DAG，描述了多个节点的先后运行顺序。

apiVersion: 工作流格式的版本
metadata:
  name: 工作流名
  identifier: 工作流标识
  version: 工作流版本
  provider: 工作流提供方
  displayName: 工作流显示名
  annotations: 工作流注解，目前只支持官方注解
spec:
  inputs: 输入定义
    artifacts: 
    parameters:
  outputs: 输出定义
    artifacts:
    parameters:
  pipelines: 子工作流集合
  - metadata:
      name: 子工作流名
      displayName: 子工作流显示名
      guid: 只用在临时节点中，代表临时节点的Id
      provider: 子工作流提供方
      identifier: 子工作流所属定义的唯一标识
      version: 子工作流所属定义的版本
    spec:
      arguments: 子工作流赋值
      	artifacts: 
    	parameters:
      dependencies: 子工作流依赖列表
      when: 条件判断
      withSequence: 基于序号的循环表达式
        start: 开始序号
        end: 结束序号
        format: 格式,具体见https://pkg.go.dev/fmt  
      withItems: 基于列表的循环表达式
      withParam: 基于参数的循环表达式
      parallelism: 节点并行度

详细定义

API版本、元信息以及输入输出定义等详情，请参见公共格式。

节点集合详情如下。

spec:
  pipelines: 节点集合
  - metadata:
      name: 节点名
      displayName: 节点显示名
      guid: 只用在临时节点中，代表临时节点的Id
      provider: 节点提供方
      identifier: 节点标识
      version: 节点版本
    spec:
      arguments: 节点赋值
      	artifacts: 节点Artifact赋值
    		parameters: 节点参数赋值
      dependencies: 节点依赖
      when: 条件判断
      withSequence: 基于序号的循环表达式
        start: 开始序号
        end: 结束序号
        format: 格式,具体见https://pkg.go.dev/fmt  
      withItems: 基于列表的循环表达式
      withParam: 基于参数的循环表达式
      parallelism: 节点并行度

Pipelines是节点集合，集合中每个节点必须先注册到PAIFlow。每个节点可以包含更多的节点集合，没有层数限制。每个节点定义如下：

示例

• DAG工作流

  工作流由三个节点构成：读数据源、类型转换、SQL转换

  • 读数据源进行数据源生成，产出一个输出Artifact，代表生成的MaxCompute表。

  • 类型转换对读数据源产出的MaxCompute表数据进行类型转换，产出转换后的MaxCompute表。

  • SQL转换对类型转换产出的MaxCompute表数据进行SQL转换，产出转换后的MaxCompute表。

  

  代码示例（单击查看详情）

  apiVersion: "core/v1"
  metadata:
    provider: "11577*******4901"
    version: "v1"
    identifier: "job-root-pipeline-identifier"
  spec:
    inputs:
      parameters:
      - name: "execution_maxcompute"
        type: "Map"
    pipelines:
    - apiVersion: "core/v1"
      metadata:
        provider: "pai"
        version: "v1"
        identifier: "data_source"
        name: "data_source"
        displayName: "读数据源"
      spec:
        arguments:
          parameters:
          - name: "inputTableName"
            value: "pai_online_project.wumai_data"
          - name: "execution"
            from: "{{inputs.parameters.execution_maxcompute}}"
    - apiVersion: "core/v1"
      metadata:
        provider: "pai"
        version: "v1"
        identifier: "type_transform"
        name: "type_transform"
        displayName: "类型转换"
      spec:
        arguments:
          artifacts:
          - name: "inputTable"
            from: "{{pipelines.data_source.outputs.artifacts.outputTable}}"
          parameters:
          - name: "cols_to_double"
            value: "time,hour,pm2,pm10,so2,co,no2"
          - name: "default_int_value"
            value: "0"
          - name: "reserveOldFeat"
            value: "false"
          - name: "execution"
            from: "{{inputs.parameters.execution_maxcompute}}"
        dependencies:
        - "data_source"
    - apiVersion: "core/v1"
      metadata:
        provider: "pai"
        version: "v1"
        identifier: "sql"
        name: "sql"
        displayName: "SQL转换"
      spec:
        arguments:
          artifacts:
          - name: "inputTable1"
            from: "{{pipelines.type_transform.outputs.artifacts.outputTable}}"
          parameters:
          - name: "sql"
            value: "select time,hour,(case when pm2>200 then 1 else 0 end),pm10,so2,co,no2\
              \ from ${t1}"
          - name: "execution"
            from: "{{inputs.parameters.execution_maxcompute}}"
        dependencies:
        - "type_transform"

• 条件表达式

  工作流由三个节点构成：读数据源、类型转换pm10、类型转换pm2

  • 读数据源进行数据源生成，产出一个输出Artifact，代表生成的MaxCompute表。

  • 如果读数据源成功，运行类型转换pm2对输入的MaxCompute表数据pm2这列进行类型转换，产出转换后的MaxCompute表。

  • 如果读数据源失败，运行类型转换pm10对输入的MaxCompute表数据pm10这列进行类型转换，产出转换后的MaxCompute表。

  

  代码示例（单击查看详情）

  apiVersion: "core/v1"
  metadata:
    provider: "1157******4901"
    version: "v1"
    identifier: "job-root-pipeline-identifier"
  spec:
    inputs:
      parameters:
      - name: "execution_maxcompute"
        type: "Map"
    pipelines:
    - apiVersion: "core/v1"
      metadata:
        provider: "pai"
        version: "v1"
        identifier: "data_source"
        name: "data_source"
        displayName: "读数据源"
      spec:
        arguments:
          parameters:
          - name: "inputTableName"
            value: "pai_online_project.wumai_data"
          - name: "execution"
            from: "{{inputs.parameters.execution_maxcompute}}"
    - apiVersion: "core/v1"
      metadata:
        provider: "pai"
        version: "v1"
        identifier: "type_transform"
        name: "type_transform_pm10"
        displayName: "类型转换pm10"
      spec:
        when: '{{pipelines.dataSource.status}} == Failed'
        arguments:
          artifacts:
          - name: "inputTable"
            from: "{{pipelines.data_source.outputs.artifacts.outputTable}}"
          parameters:
          - name: "cols_to_double"
            value: "pm10"
          - name: "default_int_value"
            value: "0"
          - name: "reserveOldFeat"
            value: "false"
          - name: "execution"
            from: "{{inputs.parameters.execution_maxcompute}}"
        dependencies:
        - "data_source"
    - apiVersion: "core/v1"
      metadata:
        provider: "pai"
        version: "v1"
        identifier: "type_transform"
        name: "type_transform_pm2"
        displayName: "类型转换pm2"
      spec:
        when: '{{pipelines.dataSource.status}} == Succeeded'
        arguments:
          artifacts:
          - name: "inputTable"
            from: "{{pipelines.data_source.outputs.artifacts.outputTable}}"
          parameters:
          - name: "cols_to_double"
            value: "pm2"
          - name: "default_int_value"
            value: "0"
          - name: "reserveOldFeat"
            value: "false"
          - name: "execution"
            from: "{{inputs.parameters.execution_maxcompute}}"
        dependencies:
        - "data_source"

• 循环表达式

  工作流由2个节点构成：echo、date_consumer，运行时会变为4个节点。

  • echo有一个输出参数，值为一个数组["a","b","c"]

  • 根据echo节点的输出参数值["a","b","c"]，date_consumer循环三次，生成三个节点。

    • 节点1名字为date_consumer(0,"a")，data为"a"。

    • 节点2名字为date_consumer(1,"b")，data为"b"。

    • 节点3名字为date_consumer(2,"c")，data为"c"。

  

  代码示例（单击查看详情）

  apiVersion: "core/v1beta1"
  metadata:
    name: "echo_repeat"
    provider: "15577*****4904"
    identifier: "forcast"
    version: "v1"
  spec:
    pipelines:
    #echo有输出参数num，值为["a","b","c"]
    - apiVersion: "core/v1"
      metadata:
        provider: "15577*****4904"
        version: "v1"
        identifier: "echo"
        name: "echo"
    #date_consumer循环3次
    - apiVersion: "core/v1"
      metadata:
        provider: "15577*****4904"
        version: "v1"
        identifier: "date_consumer"
        name: "date_consumer"
      spec:
        arguments:
          parameters:
          - name: "data"
            value: "{{item.num}}"
        withParam: "{{pipelines.echo.outputs.parameters}}"
        dependencies: "echo"

• OSS输入+自定义镜像+DLC任务

  工作流由4个节点构成：读OSS数据-1、数据转tfrecord-1、Python脚本-1、图像自监督训练-1，运行时会变为6个节点。

  • 读OSS数据-1读取OSS数据。

  • 数据转tfrecord-1把OSS数据转为tfrecord。

  • 图像自监督训练-1创建一个DLC任务，进行图像训练。

  • Python脚本-1会根据输入参数withSequence循环三次，生成三个节点；Python脚本-1使用了自定义镜像运行DLC任务。

    • 节点1名字为Python脚本-1(0,1)

    • 节点2名字为Python脚本-1(1,2)

    • 节点3名字为Python脚本-1(2,3)

  

  代码示例（单击查看详情）

  apiVersion: "core/v1"
  metadata:
    provider: "11577*****4901"
    version: "v1"
    identifier: "oss-dlc"
  spec:
    inputs:
      parameters:
      - name: "execution_maxcompute"
        type: "Map"
      - name: "execution_maxcompute_optional"
        type: "Map"
      - name: "execution_dlc_optional"
        type: "Map"
      - name: "execution_dlc"
        type: "Map"
    pipelines:
    - apiVersion: "core/v1"
      metadata:
        provider: "pai"
        version: "v1"
        identifier: "tfReadFileData"
        name: "id-26425a11-939f-4c9e-b261"
        displayName: "读OSS数据-1"
      spec:
        arguments:
          parameters:
          - name: "ossBucket"
            value: "oss://pai-vision-data-sh.oss-cn-shanghai.aliyuncs.com/data/demo_image_match/meta/train_crop_label_lt_10k_nolabel.txt"
          - name: "arn"
            value: "acs:ram::1157******94901:role/aliyunodpspaidefaultrole"
          - name: "execution"
            from: "{{inputs.parameters.execution_maxcompute}}"
    - apiVersion: "core/v1"
      metadata:
        provider: "pai"
        version: "v1"
        identifier: "ev_ssl"
        name: "id-548b1789-3acc-4874-a06b"
        displayName: "图像自监督训练-1"
      spec:
        arguments:
          artifacts:
          - name: "input_train_data"
            from: "{{pipelines.id-97c4cd39-b8df-4456-9ba6.outputs.artifacts.output_train_data}}"
          parameters:
          - name: "model_type"
            value: "MOBY_TIMM_TFRECORD_OSS"
          - name: "arn"
            value: "acs:ram::11577*****4901:role/aliyunodpspaidefaultrole"
          - name: "model_dir"
            value: "oss://alink-test-2.oss-cn-shanghai-internal.aliyuncs.com/${pai_system_run_id}/${pai_system_node_id}/"
          - name: "use_pretrained_model"
            value: "false"
          - name: "backbone"
            value: "resnet50"
          - name: "optimizer"
            value: "AdamW"
          - name: "initial_learning_rate"
            value: "0.001"
          - name: "train_batch_size"
            value: "64"
          - name: "num_epochs"
            value: "100"
          - name: "save_checkpoint_epochs"
            value: "10"
          - name: "train_num_readers"
            value: "4"
          - name: "fp16"
            value: "false"
          - name: "singleOrDistribute"
            value: "distribute_dlc"
          - name: "count"
            value: "1"
          - name: "gpuMachineType"
            value: "ecs.gn6e-c12g1.12xlarge"
          - name: "execution"
            from: "{{inputs.parameters.execution_dlc}}"
        dependencies:
        - "id-97c4cd39-b8df-4456-9ba6"
    - apiVersion: "core/v1"
      metadata:
        provider: "pai"
        version: "v1"
        identifier: "ev_convert"
        name: "id-97c4cd39-b8df-4456-9ba6"
        displayName: "数据转tfrecord-1"
      spec:
        arguments:
          artifacts:
          - name: "input_label_file"
            from: "{{pipelines.id-26425a11-939f-4c9e-b261.outputs.artifacts.output_1}}"
          parameters:
          - name: "arn"
            value: "acs:ram::11577*****4901:role/aliyunodpspaidefaultrole"
          - name: "output_dir"
            value: "oss://alink-test-2.oss-cn-shanghai-internal.aliyuncs.com/${pai_system_run_id}/${pai_system_node_id}/"
          - name: "output_prefix"
            value: "tx"
          - name: "model_type"
            value: "CLASSIFICATION"
          - name: "class_list_file"
            value: "oss://pai-vision-data-sh.oss-cn-shanghai.aliyuncs.com/data/demo_image_match/meta/test.config"
          - name: "test_ratio"
            value: "0.0"
          - name: "converter_class"
            value: "SelfSupervisedConverter"
          - name: "image_format"
            value: "jpg"
          - name: "read_parallel_num"
            value: "10"
          - name: "write_parallel_num"
            value: "1"
          - name: "num_samples_per_tfrecord"
            value: "1000"
          - name: "count"
            value: "5"
          - name: "cpu"
            value: "800"
          - name: "memory"
            value: "20000"
          - name: "execution"
            from: "{{inputs.parameters.execution_maxcompute}}"
        dependencies:
        - "id-26425a11-939f-4c9e-b261"
    - apiVersion: "core/v1"
      metadata:
        provider: "pai"
        version: "v1"
        identifier: "python_v2"
        name: "id-2e2c22ad-134f-4010-926f"
        displayName: "Python脚本V2-1"
      spec:
        withSequence:
          start: 1
          end: 3
        arguments:
          artifacts:
          - name: "input2"
            metadata:
              type:
                DataSet:
                  locationType: "OSS"
            from: "{{pipelines.id-26425a11-939f-4c9e-b261.outputs.artifacts.output_1}}"
          parameters:
          - name: "roleArn"
            value: "acs:ram::11577*****94901:role/aliyunodpspaidefaultrole"
          - name: "codeUri"
            value: "oss://alink-test-2.oss-cn-shanghai-internal.aliyuncs.com/a.py"
          - name: "_codeSourceAdvanced"
            value: "false"
          - name: "resourceId"
            value: "public-cluster"
          - name: "instanceType"
            value: "ecs.c6.large"
          - name: "_runConfigAdvanced"
            value: "true"
          - name: "imageUri"
            value: "registry.cn-shanghai.aliyuncs.com/pai-dlc/tensorflow-training:1.12PAI-cpu-py27-ubuntu18.04"
          - name: "jobType"
            value: "TFJob"
          - name: "execution_dlc"
            from: "{{inputs.parameters.execution_dlc_optional}}"
          - name: "execution_maxcompute"
            from: "{{inputs.parameters.execution_maxcompute_optional}}"
        dependencies:
        - "id-26425a11-939f-4c9e-b261"