将顶层DAG连接在一起

从中获取提示 @维拉杰·帕雷克 的 answer ,我能够 TriggerDagRunOperator 以预定的方式工作。我在此张贴我的(部分)回答;将在事情变得清楚时更新。

如何克服 parent_id 前缀输入 dag_id 属于 SubDag S?

正如@viraj所说,没有直接的方法可以做到这一点。延伸 SubDagOperator 删除 this check 可能有用,但我决定避开它

如何强迫 触发器 S至 等待DAG完成 是吗?

• 看着 implementation ,很明显 触发器 只是为了 触发 外部DAG;就这样。默认情况下,它不是 应该是 等待DAG完成。因此我观察到的行为是可以理解的。

• ExternalTaskSensor 是显而易见的出路。然而,在学习 Airflow 我在依赖 手动触发 达格的( schedule_interval=None )中。在这种情况下, 外部任务传感器 很难准确说明 execution_date 对于外部任务(正在等待谁完成),如果失败, sensor gets stuck 是的。

• 所以接受暗示 implementation ,我做的 minor adjustment 行为 外部任务传感器 等待完成 全部的 task_instance S公司 有关任务的

  execution_date[external_task] >= execution_date[TriggerDagRunOperator] + execution_delta

  这就实现了 desired result :外部DAG按顺序依次运行。

有没有办法让我 创建单独的文件 (对于仅在输入上不同的DAG)对于每个顶级DAG?

再次通过@viraj这可以通过 assigning DAGs to global scope 使用 globals()[dag_id] = DAG(..)

编辑-1

也许我指的是不正确的资源 link 上面已经死了),但是 ExternalTaskSensor 已经包含参数 execution_delta &安培 execution_date_fn 容易限制 执行日期 (s)任务是 感觉到 是的。

• 你能给出更多细节,你的意思是等待完成的DAG之前,被触发?能不能让triggerdagrunoperator成为dag中的最后一个任务?

• 要创建类似的dag,可以从一个python文件动态生成dag。你可以这样做:

从气流导入DAG

from airflow.operators.python_operator import PythonOperator


def create_dag(dag_id,
               schedule,
               dag_number,
               default_args):

def hello_world_py(*args):
    print('Hello World')
    print('This is DAG: {}'.format(str(dag_number)))

dag = DAG(dag_id,
          schedule_interval=schedule,
          default_args=default_args)

with dag:
    t1 = PythonOperator(
        task_id='hello_world',
        python_callable=hello_world_py,
        dag_number=dag_number)

return dag


# build a dag for each number in range(10)
for n in range(1, 10):
dag_id = 'hello_world_{}'.format(str(n))

default_args = {'owner': 'airflow',
                'start_date': datetime(2018, 1, 1)
                }

schedule = '@daily'

dag_number = n

globals()[dag_id] = create_dag(dag_id,
                              schedule,
                              dag_number,
                              default_args)

你可以在这里阅读更多关于这种方法的内容。如果大多数正在生成的DAG非常相似,您可能需要考虑将配置存储在一个airlow变量中 enter link description here

您可能无法克服subdag操作符的前缀限制-我建议将子dag完全从工作流中删除,并将其作为单独的dag运行-如果您发现自己必须这样做,则可以更轻松地返回并重新运行旧的dag运行。