3

假设我有如下定义的以下虚拟 DAG：

@dag(default_args=default_args,
     schedule_interval=None,
     start_date=days_ago(2))
def airflow_taskflow_api_dag():
    cur_day = '2020-01-01'

    @task()
    def A(current_day: str):
        return current_day + ' 10'

    @task(multiple_outputs=True)
    def B(current_date_time: str):
        timestamp = calendar.timegm(
            datetime.strptime(
                current_date_time, '%Y-%m-%d %H').timetuple())

        another_current_date_time_format = '2020-01-01T10:00:00'

        return {
            "timestamp": timestamp,
            "new_cur_date_time": another_current_date_time_format
        }

    @task()
    def C(livy_batch_id, start_time: str):
        logging.info("Executing task C {}".format(start_time))
       return "c_id"

    @task()
    def D(upstream_id, start_time: str):
        logging.info("Executing task D {}".format(start_time))

    # workflow dependencies
    get_current_time = A(cur_day)
    new_date_time = B(get_current_time)

    livy_task = LivyOperator(
        task_id='livy_task',
        file="dummy_file",
        class_name="dummy_class",
        args=[
            new_date_time['timestamp']
        ],
        driver_memory='1G',
        driver_cores='4',
        executor_memory='1G',
        executor_cores='4',
        num_executors='4',
        queue='default_queue',
        livy_conn_id='livy_conn_id'
    )

    c_task = C(livy_task.output, new_date_time['new_cur_date_time'])

    d_task = D(c_task, new_date_time['new_cur_date_time'])

airflow_taskflow_api_dag = airflow_taskflow_api_dag()

如果我这样编写我的 DAG（从上游任务传递两个参数），则依赖项将类似于： 屏幕截图 2021-03-11 下午 3 17 37

但我想定义依赖： A >> B >> livy_task >> C >> D

有没有办法通过使用taskflow api来做到这一点？似乎如果我将任务 B 的输出传递给 livy_task、C 和 D，这三个任务将并行运行。

谢谢！

4

0 回答 0