创建和运行工作流

创建一个新的工作流主要包括以下流程：

1. 定义工作流的输入信息。

   输入信息包括用户输入参数PipelineParameter或数据输入PipelineArtifact。对于数据输入，目前PAI的工作流服务支持OSS数据、MaxCompute表数据及MaxCompute OfflineModel。

2. 创建工作流中的Step及其输入。

   Step的输入可能来自于其他Step，也可能来源于当前创建的工作流的输入。

3. 指定工作流的输出信息。

   工作流可以使用引用的方式将Step节点的输出作为新的工作流的输出。

下面的示例代码中，构建了一个简单的工作流，它先使用类型转换组件将输入表的部分字段转换为浮点类型，再通过数据拆分组件将表拆分为两张MaxCompute表。

from pai.pipeline.types import PipelineParameter, PipelineArtifact, ArtifactMetadataUtils
from pai.pipeline import PipelineStep, Pipeline, RegisteredComponent

def create_composite_pipeline():
    # 定义当前工作流的输入信息。
    
    # 由于组件底层的处理是运行在MaxCompute上，execution参数传递执行的MaxCompute Project信息。
    execution_input = PipelineParameter(name="execution", typ=dict)
    cols_to_double_input = PipelineParameter(name="cols_to_double")
    table_input = PipelineArtifact(name="input_table", metadata=ArtifactMetadataUtils.maxc_table())

    # 构建类型转换Step。
    # 指定identifier-provider-version, 使用一个已经保存的组件，作为工作流的一个Step。
    type_transform_step = PipelineStep.from_registered_component(
        identifier="type_transform", provider=ProviderAlibabaPAI,
        version="v1", name="typeTransform", inputs={
            "inputTable": table_input,
            "execution": execution_input,
            # "outputTable": gen_temp_table(), 
            "cols_to_double": cols_to_double_input,
        }
    )
    
    # 构建拆分表Step。
    # SavedOperator也可以作为一个Step构建工作流。
    split_operator = RegisteredComponent.get_by_identifier(identifier="split",
     provider=ProviderAlibabaPAI, version="v1")
    split_step = split_operator.as_step(inputs={"inputTable": type_transform_step.outputs[0],
            "execution": execution_input,
            # "output1TableName": gen_temp_table(),
            "fraction": 0.5,
            # "output2TableName": gen_temp_table(),
        })

    # Pipeline构造函数中的steps和inputs信息并不要求完整输入，Pipeline graph时，是通过Pipeline的outputs和steps推导他们的依赖，从而构造对应的执行DAG。
    p = Pipeline(
        steps=[split_step],
        outputs=split_step.outputs[:2],
    )
    return p

p = create_composite_pipeline()
# 输入工作流运行所需参数(arguments）后，提交到PAI Pipeline Service运行。
pipeline_run = p.run(job_name="demo-composite-pipeline-run", arguments={
            "execution": maxc_execution,
            "cols_to_double": "time,hour,pm2,pm10,so2,co,no2",
            "input_table": "odps://pai_online_project/tables/wumai_data",
        }, wait=True)


pipeline_run.get_outputs()

对于上述构建好的工作流，您可以为其指定组件名称和版本，将该工作流保存到服务端成为一个可复用组件。保存的组件默认共享给阿里云账号下的所有RAM用户，保存组件的示例代码如下。

# 指定identifier和版本，保存工作流。保存的组件的provider默认为对应的阿里云账号的UID。
p = p.save(identifier="demo-composite-pipeline", version="v1")
print(p.pipeline_id, p.identifier, p.version, p.provider)

您可以在本地的Python环境或DSW中，完成SDK初始化（参见安装和配置）后，运行以下的示例代码训练心脏病预测模型。为了保存输出的PMML模型文件，工作流在运行时，需要提供您的OSS信息。

import random
from pai.pipeline.types import (
    ArtifactMetadataUtils,
    ParameterType,
    PipelineArtifact,
    PipelineParameter,
)
from pai.pipeline import Pipeline, PipelineStep

feature_cols = "sex,cp,fbs,restecg,exang,slop,thal,age,trestbps,chol,thalach,oldpeak,ca,ifhealth"
label_col = "ifhealth"

def create_pipeline():
    pmml_oss_bucket = PipelineParameter("pmml_oss_bucket")
    pmml_oss_rolearn = PipelineParameter("pmml_oss_rolearn")
    pmml_oss_path = PipelineParameter("pmml_oss_path")
    pmml_oss_endpoint = PipelineParameter("pmml_oss_endpoint")
    execution = PipelineParameter("execution", ParameterType.Map)
    dataset_input = PipelineArtifact(
        "dataset-table",
        metadata=ArtifactMetadataUtils.maxc_table(),
        required=True,
    )

    sql = (
        "select age, (case sex when 'male' then 1 else 0 end) as sex,(case cp when"
        " 'angina' then 0  when 'notang' then 1 else 2 end) as cp, trestbps, chol,"
        " (case fbs when 'true' then 1 else 0 end) as fbs, (case restecg when 'norm'"
        " then 0  when 'abn' then 1 else 2 end) as restecg, thalach, (case exang when"
        " 'true' then 1 else 0 end) as exang, oldpeak, (case slop when 'up' then 0  "
        "when 'flat' then 1 else 2 end) as slop, ca, (case thal when 'norm' then 0 "
        " when 'fix' then 1 else 2 end) as thal, (case status when 'sick' then 1 else "
        "0 end) as ifHealth from ${t1};"
    )
    sql_step = PipelineStep.from_registered_component(
        "sql",
        name="sql-1",
        provider=ProviderAlibabaPAI,
        version="v1",
        inputs={
            "inputTable1": dataset_input,
            "execution": execution,
            "sql": sql,
        },
    )

    type_transform_step = PipelineStep.from_registered_component(
        "type_transform",
        name="type-transform-1",
        provider=ProviderAlibabaPAI,
        version="v1",
        inputs={
            "execution": execution,
            "inputTable": sql_step.outputs["outputTable"],
            "cols_to_double": feature_cols,
            # "outputTable": gen_run_node_scoped_placeholder(suffix="outputTable"),
        },
    )

    normalize_step = PipelineStep.from_registered_component(
        "normalize_1",
        name="normalize-1",
        provider=ProviderAlibabaPAI,
        version="v1",
        inputs={
            "execution": execution,
            "inputTable": type_transform_step.outputs["outputTable"],
            "selectedColNames": feature_cols,
            "lifecycle": 1,
            # "outputTableName": gen_run_node_scoped_placeholder(suffix="outputTable"),
            # "outputParaTableName": gen_run_node_scoped_placeholder(
            #     suffix="outputParaTable"
            # ),
        },
    )

    split_step = PipelineStep.from_registered_component(
        identifier="split",
        name="split-1",
        provider=ProviderAlibabaPAI,
        version="v1",
        inputs={
            "inputTable": normalize_step.outputs["outputTable"],
            "execution": execution,
            "fraction": 0.8,
        },
    )

    model_name = "test_health_prediction_by_pipeline_%s" % (random.randint(0, 999999))

    lr_step = PipelineStep.from_registered_component(
        identifier="logisticregression_binary",
        name="logisticregression-1",
        provider=ProviderAlibabaPAI,
        version="v1",
        inputs={
            "inputTable": split_step.outputs["output1Table"],
            "execution": execution,
            "generatePmml": True,
            "pmmlOssEndpoint": pmml_oss_endpoint,
            "pmmlOssBucket": pmml_oss_bucket,
            "pmmlOssPath": pmml_oss_path,
            "pmmlOverwrite": True,
            "roleArn": pmml_oss_rolearn,
            "regularizedLevel": 1.0,
            "regularizedType": "l2",
            "modelName": model_name,
            "goodValue": 1,
            "featureColNames": ",".join(feature_cols),
            "labelColName": label_col,
        },
    )

    offline_model_pred_step = PipelineStep.from_registered_component(
        identifier="Prediction_1",
        name="offlinemodel-pred",
        provider=ProviderAlibabaPAI,
        version="v1",
        inputs={
            "model": lr_step.outputs["model"],
            "inputTable": split_step.outputs["output2Table"],
            "execution": execution,
            "featureColNames": feature_cols,
            "appendColNames": label_col,
        },
    )

    evaluate_step = PipelineStep.from_registered_component(
        identifier="evaluate_1",
        name="evaluate-1",
        provider=ProviderAlibabaPAI,
        version="v1",
        inputs={
            "execution": execution,
            "inputTable": offline_model_pred_step.outputs["outputTable"],
            "scoreColName": "prediction_score",
            "labelColName": label_col,
        },
    )

    p = Pipeline(
        steps=[evaluate_step, offline_model_pred_step],
        outputs={
            "pmmlModel": lr_step.outputs["PMMLOutput"],
            "evaluateResult": evaluate_step.outputs["outputMetricTable"],
        },
    )
    return p

p = create_pipeline()

# 输出Pipeline的图片。
p.dot().render()

pmml_oss_endpoint = "<YourOssBucketEndpoint>"
pmml_oss_path = "<PathForThePmmlFile>"
pmml_oss_bucket = "<YourOssBucketName>"

pmml_oss_rolearn = "<RoleArnForPaiToVisitOssBucket>"
maxc_execution = {
    "endpoint": "<your_Project_Endpoint>",
    "odpsProject": "<your_MaxCompute_project>",
}

run_instance = p.run(
    job_name="test_heart_disease_pred",
    arguments={
        "execution": maxc_execution,
        "pmml_oss_rolearn": pmml_oss_rolearn,
        "pmml_oss_path": pmml_oss_path,
        "pmml_oss_bucket": pmml_oss_bucket,
        "pmml_oss_endpoint": pmml_oss_endpoint,
        "inputTable": "odps://pai_online_project/tables/heart_disease_prediction"
    },
    wait=True,
)
print(run_instance.get_outputs())

需要根据实际情况，替换以下参数值。

上述代码构建的工作流，通过pipeline.dot().render()渲染后，得到的Pipeline DAG如下图所示。