0

我创建了一个 DAG，为每个工作流安排 @daily 间隔和单独的任务 ID。但它并没有像例外那样运行。有可能这样做吗？有没有其他方法可以为特定的 dag 创建动态任务？并使用命令行暂停特定任务实例？

from __future__ import print_function
from builtins import range
from airflow.operators import PythonOperator,DummyOperator,BranchPythonOperator,SqlSensor
from airflow.models import DAG
from datetime import datetime, timedelta

import time
from pprint import pprint

seven_days_ago = datetime.combine(
        datetime.today() - timedelta(7), datetime.min.time())

args = {
    'owner': 'varakumar',
    'start_date': seven_days_ago,
}

dag = DAG(
    dag_id='dynamic_task_creation', default_args=args,
    schedule_interval="@daily")

def get_decision():
    return "right"

start = DummyOperator(
    task_id='start',
    dag=dag)

td=datetime.today()
x=str(datetime(td.year,td.month,td.day,td.hour,td.minute,td.second)).replace (" ", "_").replace (":", "-")
pause_task_id = ("pause-%s" % x)

pause = DummyOperator(
    task_id=pause_task_id,
    dag=dag)
pause.set_upstream(start)

decision = BranchPythonOperator(
    task_id='decision',
    python_callable=lambda: get_decision(),
    dag=dag)
decision.set_upstream(pause)

left = DummyOperator(
    task_id='left',
    dag=dag)
left.set_upstream(decision)

right = DummyOperator(
    task_id='right',
    dag=dag)
right.set_upstream(decision)

先感谢您

4

1 回答 1

0

我看到的第一个问题是您使用的是动态start_date. 我在这样做时看到了一些奇怪的行为，我认为这是基于气流维护其过去 dagruns 列表的方式。尝试指定一个固定值start_date，看看是否能解决任何问题。

在任何一种情况下，气流文档都建议不要使用动态开始日期（向下滚动并阅读说明start_date）。

编辑：另请查看此内容以获取更多信息。

于 2016-09-08T22:12:23.920 回答