如何将资源传递给Dagster Job供内部Op访问并解决配置缺失报错

AI 小北

2026-9-25

报错原因

你定义的boto3_connection资源显式声明了两个必填配置项aws_access_key_id和aws_secret_access_key，但调用Job运行时没有传入对应的资源配置，因此触发配置校验错误。

解决方案

方案1：运行Job时传入配置

本地直接运行Job时，调用my_job.execute_in_process()传入run_config参数即可，示例如下：

if __name__ == "__main__":
    my_job.execute_in_process(
        run_config={
            "resources": {
                "boto3_connection": {
                    "config": {
                        "aws_access_key_id": os.getenv("AWS_ACCESS_KEY_ID"),
                        "aws_secret_access_key": os.getenv("AWS_SECRET_ACCESS_KEY")
                    }
                }
            }
        }
    )

如果是在Dagster UI中触发运行，在运行配置页填写对应层级的resources配置项即可。

方案2：资源默认读取环境变量，无需每次传配置

如果不想每次运行都手动传配置，可以修改资源定义，直接从环境变量读取值，不需要额外声明必填配置：

@resource
def boto3_connection(context):
    return Boto3Connector(
        aws_access_key_id=os.getenv("AWS_ACCESS_KEY_ID"),
        aws_secret_access_key=os.getenv("AWS_SECRET_ACCESS_KEY")
    )

修改后运行Job不需要再额外传入run_config参数即可直接运行。