YAML 焦虑再见：PythonSDK 助力大规模 Argo Workflows 构建

pip install hera-workflows

# 导入相关包from hera.workflows import DAG, Workflow, scriptfrom hera.shared import global_configimport urllib3
urllib3.disable_warnings()
# 配置访问地址和tokenglobal_config.host = "https://argo.{{clusterid}}.{{region-id}}.alicontainer.com:2746"global_config.token = "abcdefgxxxxxx" # 填入之前获取的tokenglobal_config.verify_ssl = ""
# 装饰器函数script是 Hera 实现近乎原生的 Python 函数编排的关键功能。# 它允许您在 Hera 上下文管理器（例如Workflow或Steps上下文）下调用该函数，# 该函数在任何 Hera 上下文之外仍将正常运行，这意味着您可以在给定函数上编写单元测试。# 该示例是打印输入的信息。@script()def echo(message: str):    print(message)
# 构建workflow，Workflow是 Argo 中的主要资源，# 也是 Hera 的关键类，负责保存模板、设置入口点和运行模板。with Workflow(    generate_name="dag-diamond-",    entrypoint="diamond",) as w:    with DAG(name="diamond"):        A = echo(name="A", arguments={"message": "A"}) # 构建template        B = echo(name="B", arguments={"message": "B"})        C = echo(name="C", arguments={"message": "C"})        D = echo(name="D", arguments={"message": "D"})        A >> [B, C] >> D # 构建依赖关系，B、C任务依赖A，D依赖B和C# 创建workfloww.create()

python simpleDAG.py

from hera.workflows import DAG, Artifact, NoneArchiveStrategy, Parameter, OSSArtifact, Workflow, scriptfrom hera.shared import global_config


    
import urllib3
urllib3.disable_warnings()# 设置访问地址global_config.host = "https://argo.{{clusterid}}.{{region-id}}.alicontainer.com:2746"global_config.token = "abcdefgxxxxxx" # 填入之前获取的tokenglobal_config.verify_ssl = ""
# 使用script装饰函数时，将script参数传递给script装饰器。这包括image、inputs、outputs、resources等。@script(    image="python:alpine3.6",    inputs=Parameter(name="num_parts"),    outputs=OSSArtifact(name="parts", path="/mnt/out", archive=NoneArchiveStrategy(), key="{{workflow.name}}/parts"),)def split(num_parts: int) -> None: # 根据输入参数num_parts创建多个文件，文件中写入foo字符和parts编号    import json    import os    import sys
    os.mkdir("/mnt/out")
    part_ids = list(map(lambda x: str(x), range(num_parts)))    for i, part_id in enumerate(part_ids, start=1):        with open("/mnt/out/" + part_id + ".json", "w") as f:            json.dump({"foo": i}, f)    json.dump(part_ids, sys.stdout)
# script中定义image、inputs、outputs@script(    image="python:alpine3.6",    inputs=[Parameter(name="part_id", value="0"), Artifact(name="part", path="/mnt/in/part.json"),],    outputs=OSSArtifact(        name="part",        path="/mnt/out/part.json",        archive=NoneArchiveStrategy(),        key="{{workflow.name}}/results/{{inputs.parameters.part_id}}.json",    ),)def map_() -> None: # 根据文件中foo字符的个数，生成新文件，将foo内容parts编号乘以2，写入bar内容    import json    import os
    os.mkdir("/mnt/out")    with open("/mnt/in/part.json") as f:        part = json.load(f)    with open("/mnt/out/part.json", "w") as f:        json.dump({"bar": part["foo"] * 2}, f)
# script中定义image、inputs、outputs、resources@script(    image="python:alpine3.6",    inputs=OSSArtifact(name="results", path="/mnt/in", key="{{workflow.name}}/results"),    outputs=OSSArtifact(        name="total", path="/mnt/out/total.json", archive=NoneArchiveStrategy(), key="{{workflow.name}}/total.json"    ),)def reduce() -> None: # 计算每个parts的bar的值的总和。    import json    import os
    os.mkdir("/mnt/out")
    total = 0    for f in list(map(lambda x: open("/mnt/in/" + x), os.listdir("/mnt/in"))):        result = json.load(f)        total = total + result["bar"]    with open("/mnt/out/total.json", "w") as f:        json.dump({"total": total}, f)
# 构建workflow，输入name、设置入口点、namespace、全局参数等。with


    
 Workflow(generate_name="map-reduce-", entrypoint="main", namespace="default", arguments=Parameter(name="num_parts", value="4")) as w:    with DAG(name="main"):        s = split(arguments=Parameter(name="num_parts", value="{{workflow.parameters.num_parts}}")) # 构建templetes        m = map_(            with_param=s.result,            arguments=[Parameter(name="part_id", value="{{item}}"), OSSArtifact(name="part", key="{{workflow.name}}/parts/{{item}}.json"),],        ) # 输入参数并构建templetes，        s >> m >> reduce() # 构建任务依赖关系# 创建工作流w.create()

python map-reduce.py