Есть ли какое-либо TriggerRule для оператора воздушного потока без триггера по времени (необходимо

Question

У меня есть вариант использования для создания двух задач BigqueryOperator с одинаковой целевой таблицей, но мне нужно, чтобы одна запускалась ежедневно, а вторая запускалась вручную, когда мне нужно.

Ниже приведены иллюстрации Древовидное представление

 | task_3rd_adhoc
 | task_3rd
 |---- task_2nd
        |---- task_1st_a
        |---- task_1st_b

Из приведенного выше примера DAG запускается ежедневно. И я стремлюсь к тому, чтобы задача была:

1. task_1st_a и task_1st_b запускаются первыми. Целевые таблицы:
   • project.dataset.table_1st_a с _PARTITIONTIME = дата выполнения и
   • project.dataset.table_1st_b с _PARTITIONTIME = дата выполнения.
2. тогда task_2nd_a будет выполняться после task_1st_a и task_1st_b fini sh. BigQueryOperator использует TriggerRule.ALL_SUCCESS. Целевая таблица:
   • project.dataset.table_2nd с _PARTITIONTIME = дата выполнения.
3. , тогда task_3rd будет запущена после успеха task_2nd. BigQueryOperator использует TriggerRule.ALL_SUCCESS. Целевая таблица:
   • project.dataset.table_3rd с PARTITIONTIME = D-2 с даты выполнения.
4. task_3rd_adho c не будет выполняться в ежедневном задании. Мне это нужно, когда я хочу заполнить таблицу project.dataset.table_3rd. С целевой таблицей:
   • project.dataset.table_3rd с _PARTITIONTIME = execution_date

Но я все еще не могу найти правильное TriggerRule для шага 4 выше. Я попробовал TriggerRule.DUMMY, потому что думал, что его можно использовать, чтобы не устанавливать триггер, но task_3rd_adho c также запускается в ежедневном задании, когда я пытался создать DAG выше. (на основе this do c зависимости предназначены только для показа, запускаются по желанию)

Answer 1

Прежде всего, вы неправильно поняли TriggerRule.DUMMY.

• Обычно, когда вы соединяете задачи вместе task_a >> task_b, B запускается только после завершения A (успех / неудача, в зависимости от B trigger_rule).
• TriggerRule.DUMMY означает, что даже после соединения задач A и B вместе, как и раньше, B будет работать независимо от A ( выполняется по желанию ). Это не означает, что запускается с по вашей воле , скорее он работает с Airflow's will (он будет запускать его, когда захочет). Таким образом, очевидно, что задачи, имеющие фиктивное правило триггера, будут ВСЕГДА запускаться, хотя и в непредсказуемое время

Что вам здесь нужно (чтобы конкретная задача всегда находилась в DAG, но запускать ее только при указании вручную) - это комбинация

• AirflowSkipException
• Variable

Вот примерно то, как вы можете сделать

• A Variable должен удерживать команду для этой задачи (независимо от того, он должен работать). Эту переменную, конечно, вы можете редактировать в любое время из пользовательского интерфейса (тем самым контролируя, будет ли эта задача выполняться в следующем DagRun).
• В коде оператора (execute() метод для настраиваемого оператора или просто python_callable в случай PythonOperator), вы проверите значение переменной (должна ли задача запускаться или нет)
• На основе значения переменной, если задача НЕ должна запускаться, вы должны бросить AirflowSkipException, чтобы задача была отмечена как пропущенная . В противном случае он будет работать как обычно