如何调用 dag 任务? [英] How to call a dag task?

问题描述

我有两个 dag，分别是 Dag 1 和 Dag 2

I have two dags Dag 1 and Dag 2

任务 A、B 和 C 位于 Dag 1 中任务 D、E 和 F 位于 Dag 2

task A, B and C are in a Dag 1 task D, E and F are in Dag 2

如何实现第三个 Dag 来利用这些任务?

How to implement a third Dag to utilize the tasks?

我希望 Dag 3 使用任务 A，然后使用任务 E，然后使用任务 C，而无需重写函数

I want Dag 3 to use task A, then task E and then task C without rewriting the functions

推荐答案

创建一个返回运算符的函数，并从您需要的任何 DAG 中重用它.

Create a function that return an operator and reuse it from whatever DAG you need.

示例:

def create_my_opeartor(task_id=None, **kwargs):
    #Replace with your actual operator configuration
    return MyOperator(task_id=task_id, **kwargs) 

def create_my_opeartor_2(task_id=None, **kwargs):
    #Replace with your actual operator configuration
    return BashOperator(task_id=task_id, bash_command='echo "hello world", **kwargs)

with DAG(
    dag_id='DAG1',
    default_args=default_args,
    schedule_interval=None,
    start_date=datetime(2021, 8, 24),
) as dag1:
    a = create_my_opeartor(task_id='task_A')
    
with DAG(
    dag_id='DAG2',
    default_args=default_args,
    schedule_interval=None,
    start_date=datetime(2021, 8, 24),
) as dag2:
    e = create_my_opeartor_2(task_id='task_E')
    
with DAG(
    dag_id='DAG3',
    default_args=default_args,
    schedule_interval=None,
    start_date=datetime(2021, 8, 24),
) as dag3:
    a = create_my_opeartor(task_id='task_A')
    e = create_my_opeartor2(task_id='task_E')
    a >> e

这篇关于如何调用 dag 任务?的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！