Я использую Condor для распределенного выполнения большого количества задач обработки. Есть два этапа обработки. На первом этапе обработки я запускаю инструмент UMPTEEN раз, чтобы проанализировать некоторое подмножество исходных данных и преобразовать его в промежуточный файл. Каждое выполнение инструмента независимо от всех остальных. Итак, это хорошо подходит для использования Кондором.
Загвоздка в том, что инструмент может решить не выводить промежуточный файл. Таким образом, я не могу знать априори, сколько у меня будет промежуточных файлов; число может быть меньше UMPTEEN. Еще одна загвоздка в том, что я не знаю, каким будет имя промежуточного файла; Я знаю имя файла только после того, как оно было создано инструментом.
На втором этапе обработки я запускаю другие инструменты для преобразования каждого промежуточного файла в другие файлы назначения с другими форматами. Я хотел бы использовать Condor для этого тоже. Но чтобы написать для этого файл описания отправки, я должен точно знать, сколько промежуточных файлов мне нужно преобразовать и каковы их имена файлов.
Я попытался создать узел «generate_stage2» в моей DAG stage1, который зависит от завершения первого узла. В узле «generate_stage2» я запускаю скрипт Python, который:
- ищет промежуточные файлы;
- записывает файлы описания отправки, которые преобразуют эти промежуточные файлы в целевые форматы;
- вызывает
condor_submit_dagдля выполнения этой второй DAG.
Но отправить вторую DAG не удается. Я подозреваю, что Кондору не нравится, когда я вызываю condor_submit_dag внутри узла, который в данный момент работает в первой DAG.
Большой вопрос
Возможно ли то, что я пытаюсь? Есть ли способ, чтобы одна DAG активировала другую DAG?
Пример
Ниже приведены примеры моих файлов описания отправки, которые, надеюсь, объяснят, что я пытался сделать.
Стадия 1 ДАГ
JOB 10_src_to_int work/condor/10_src_to_int
JOB 20_generate_stage2 work/condor/20_generate_stage2
PARENT 10_src_to_int CHILD 20_generate_stage2
10_src_to_int
getenv = true
notification = Never
universe = vanilla
run_as_owner = true
initialdir = /foo/somewhere
executable = /bin/src_to_int
# UMPTEEN entries:
arguments = "src_data/ int_data/ --region -45 -123 -44 -122"
queue
arguments = "src_data/ int_data/ --region -46 -123 -45 -122"
queue
...
20_generate_stage2
getenv = true
notification = Never
universe = vanilla
run_as_owner = true
initialdir = /foo/somewhere
executable = /scripts/generate_stage2
arguments = "'data to share' 'between stage1' 'and stage2'"
queue
Стадия 2 DAG
JOB 30_int_to_dst_a work/condor/30_int_to_abc
JOB 40_int_to_dst_b work/condor/40_int_to_xyz
30_int_to_abc
# Written by the generate_stage2 script which a node in the stage1 DAG executed.
getenv = true
notification = Never
universe = vanilla
run_as_owner = true
initialdir = /foo/somewhere
executable = /bin/int_to_abc
# At most UMPTEEN entries:
arguments = "int_data/S45_W123.int out_data/S45_W123.abc"
queue
arguments = "int_data/S46_W123.int out_data/S46_W123.abc"
queue
...
40_int_to_xyz
# Written by the generate_stage2 script which a node in the stage1 DAG executed.
getenv = true
notification = Never
universe = vanilla
run_as_owner = true
initialdir = /foo/somewhere
executable = /bin/int_to_xyz
# At most UMPTEEN entries:
arguments = "int_data/S45_W123.int out_data/S45_W123.xyz"
queue
arguments = "int_data/S46_W123.int out_data/S46_W123.xyz"
queue
...
(Да, я разбиваю исходные данные на геопространственные регионы. В примерах я использовал произвольные координаты около 45° ю.ш. 123° з.д., что находится в середине океана. Это не имеет значения.)
