将参数添加到 dagster 输入

【问题标题】：Add param to dagster input将参数添加到 dagster 输入
【发布时间】：2022-06-17 00:23:50
【问题描述】：

您好，我在 dagster 中有以下 @job

@job
def job_extract_faces():
    faces = op_extract_data(op_get_data_path())
    r = op_process((faces, 'a'))
    r = op_process((faces, 'b'))
    r = op_process((faces, 'c'))
    r = op_process((faces, 'd'))

问题是 dagster 说 op_process 的输入应该是 op_extrac_data 的输出

无论如何添加参数而不是构建4个函数？

比

【问题讨论】：

标签： python dagster

【解决方案1】：

我认为您正在寻找Dynamic Graphs。使用此模式，您可以将参数作为 DynamicOutputs 从上游操作发出，并将该输出映射到 op_process。一种选择是执行以下操作：

from dagster import op, job, DynamicOut, DynamicOutput


@op(config_schema={"param_list": [str]},
    out=DynamicOut(str))
def param_generator(context):
    for i, p in enumerate(context.op_config["param_list"]):
        yield DynamicOutput(p, mapping_key=str(i))

@job
def job_extract_faces():
    faces = op_extract_data(op_get_data_path())
    param_generator().map(lambda p: op_process(faces, p))

这类似于文档中的映射示例，显示如何执行dynamic mapping with additional arguments

【讨论】：