气流的有条件调度

Question

我有两个DAG，它们在Airflow上的时间表不同，DAG1每小时在Redshift上创建一个表，DAG2是每天运行的过程，需要读取由DAG1创建的表。我面临的一个问题是，如果DAG2比平时花费更长的时间，然后DAG1开始运行，则在Redshift上出现并发连接问题。

是否有任何方法可以在DAG1中设置传感器以检查DAG2是否正在运行，并且在这种情况下要等到传感器完成后才能继续？

我尝试使用ExternalTaskSensor，但它仅检查DAG的上一次运行是否成功，而不检查DAG是否正在运行。还是我错了？

是否有任何方法可以获取DAG的状态并使用它来重新计划另一个DAG？

Answer 1

您可以让DAG2的任务失败并在10分钟后重新开始，否则您知道DAG1的任何其他合理的持续时间可能会结束。

以下设置使您的任务每10分钟重试6次，因此，如果您的每小时工作（DAG1）花费的时间不超过大约一小时，则此工作就可以了。将其添加到DAG2部分的default_args中：

default_args = {
    ...
    # Your default_args
    ...
    'retries': 6,
    'retry_delay': timedelta(minutes=10)
}

请注意不要重复写入这些属性，如果retries中已经存在retry_delay或DAG2，请覆盖它们。