Dagster调用solid添加额外参数触发DagsterInvalidDefinitionError如何修复

错误原因 #

报错是因为Dagster的pipeline构图阶段（即@pipeline装饰的函数执行阶段）不支持直接传递字面量（比如字符串、数字常量）作为solid的输入参数，所有solid的输入必须是上游solid的输出对象，或者通过配置体系传入。

修复方案 #

方案1：将固定/可配置参数改为solid配置（推荐） #

对于不需要从其他solid动态传递的参数，推荐放到solid的配置项中，而非定义为输入：

from dagster import pipeline, repository, solid, Field

@solid
def hello():
    return 1

@solid(
    # 把name、age从InputDefinition移到config_schema中
    config_schema={
        "name": Field(str),
        "age": Field(int),
    }
)
def hello2(context, hello_: int):
    # 从context中读取配置值
    name = context.solid_config["name"]
    age = context.solid_config["age"]
    print(f"Hello {name}, {age+hello_}")

@pipeline
def pipe_2():
    x = hello()
    # 调用时通过.config方法传入配置值
    hello2(x).config({
        "name": "Marcus",
        "age": 20
    })

# 注意：原代码中my_schedule未定义，这里暂时移除避免报错，可自行补全schedule定义后再加回来
@repository
def deploy_docker_repository():
    return [pipe_2]

方案2：用静态值生成solid传递字面量 #

如果确实需要保留name、age作为solid的输入（比如后续可能需要从其他solid动态传入值），可以用Dagster内置的make_values_solid生成输出静态值的solid，再把输出作为参数传递：

from dagster import pipeline, repository, solid, InputDefinition, make_values_solid

@solid
def hello():
    return 1

@solid(
    input_defs=[
        InputDefinition("hello_", int),
        InputDefinition("name", str),
        InputDefinition("age", int),
    ]
)
def hello2(hello_: int, name: str, age: int):
    print(f"Hello {name}, {age+hello_}")

# 生成输出静态值的专用solid
name_provider = make_values_solid({"name": "Marcus"})
age_provider = make_values_solid({"age": 20})

@pipeline
def pipe_2():
    x = hello()
    name_val = name_provider()
    age_val = age_provider()
    hello2(x, name_val, age_val)

@repository
def deploy_docker_repository():
    return [pipe_2]

其他注意事项 #

原代码中存在两个小问题也需要修正：

• 重复导入了solid，可删除重复的导入语句
• my_schedule未定义就放到了repository返回列表中，需要补全schedule定义或者移除

内容的提问来源于stack exchange，提问作者Alejandro A