Последовательное выполнение графических операций, не затрагивая нисходящий поток ⇐ Python

[Anonymous]

Цель состоит в том, чтобы 1) создать набор операций, которые запускают модели с заданными параметрами (реализовано с использованием фабрики операций), 2) если запуск модели не удался, должны быть запущены следующие модели (цепочка приведет к тому, что нижестоящие операции будут зависеть от вышестоящих - поэтому реализован обходной путь, установив максимальный предел параллелизма равным 1), 3) сразу после запуска каждой модели, очистите выходной каталог с помощью clean_results_dir, 4) только и только после запуска всех моделей, даже если были сбои модели, затем запустите метод операцииgener_reports.
(3) и (4) — вот где я борюсь.
Что я вижу в пользовательском интерфейсе dagster, когда задание выполняется, это странное поведение... я вижу запуск модели 1, запуск модели 2, затем cleandir1, cleandir2.... не обязательно всегда в таком порядке... но определенно не то, что мне хотелось бы делать.

Код: Выделить всё

@graph
def run_all_models_graph():
model_configs = [{"model_name":"my/model/url/","numPaths":10},{"model_name_two":"my/model/url_two/","numPaths":33}]
last_op = None
for config in model_configs:
model_name = config["model_name"]
model_name_formatted = model_name.split("/")[-1]

model_instance = get_from_op_factory(model_name_formatted, model_name, config["numPaths"])
last_op = clean_results_dir(model_instance())
return last_op

@job(
config={
"execution":{
"config": {
"multiprocess": {
"max_concurrent":1
}
}
}
}
)
def run_all_models_job():
result = run_all_models_graph()
generate_reports(start_after=result)

Подробнее здесь: https://stackoverflow.com/questions/790 ... downstream