在dagster中处理用户输入

Question

我是达格斯特的新手，我正在努力理解用户输入是如何被它处理的。我正在用下面的代码对此进行测试：

from dagster import job, op


@op
def input_string():
    ret = input('Enter string')
    print(ret)


@job
def my_job():
    input_string()


if __name__ == '__main__':
    my_job.execute_in_process()

然后在控制台中运行以下命令：

dagit -f test.py

然而，当我最终“启动运行”时，我没有机会输入输入，而是获得一个EOFError，其中包含以下信息：

dagster.core.errors.DagsterExecutionStepExecutionError:错误发生在执行op "input_string"：文件"C:\Users\username\Anaconda3\lib\site-packages\dagster\core\execution\plan\execute_plan.py"，第232行，在dagster_event_sequence_for_step for step_event in check.generator(Step_events)中:文件"C:\Users\username\Anaconda3\lib\site-packages\dagster\core\execution\plan\execute_step.py"，第354行，在_step_output_error_checked_user_event_sequence for user_event in check.generator中(文件"C:\Users\username\Anaconda3\lib\site-packages\dagster\core\execution\plan\compute.py"，行70，在_step_output_error_checked_user_event_sequence for user_event in user_event_sequence: File“C：\Users\username\Anaconda3\lib\site-packages\dagster\core\execution\plan\compute.py”，第170行，在execute_core_compute for step_output in yield_compute_results(step_context，inputs，compute_fn)中:文件step_context第138行，在yield_compute_results中为iterate_with_context中的事件(文件"C:\Users\username\Anaconda3\lib\site-packages\dagster\utils_init.py"，行403，在"C:\Users\username\Anaconda3\lib\contextlib.py"，返回文件的第137行中，在退出self.gen.throw(类型、值、跟踪)文件的第73行中，在"C:\Users\username\Anaconda3\lib\site-packages\dagster\core\execution\plan\utils.py"，solid_execution_error_boundary error_cls中(上面的异常是由以下异常引起的:在读取一行文件solid_execution_error_boundary第47行时，在solid_execution_error_boundary "C:\Users\username\Anaconda3\lib\site-packages\dagster\utils_init.py"，第401行中，EOFError: EOF，在iterate_with_context next_output =next(迭代器)文件的第65行中，在_coerce_solid_compute_fn_to_iterator next_output=fn(上下文，**kwargs)中，如果context_arg_provided _coerce_solid_compute_fn_to_iterator next_output(**kwargs)文件"test.py"，第14行，在input_string ret =input(‘输入字符串’)

我怎么才能让它运行呢？

Answer 1

ops使用配置模式进行配置。这允许您通过Dagit Launchpad提供配置。

在您的示例中，您希望从input代码中删除@op调用。然后，您将使用context.op_config字典从提供给op的config对象中检索输入，如下所示：

@op(config_schema={'input1': str})
def input_string(context):
    ret = context.op_config['input1']
    print(ret)

@job
def my_job():
    input_string()


if __name__ == '__main__':
    my_job.execute_in_process()

编辑:要让输入在Dagster作业控制台中打印，请使用内置的Dagster记录仪，如下所示：

@op(config_schema={'input1': str})
def input_string(context):
    ret = context.op_config['input1']
    context.log.info(ret)