python-2.7 - Airflow 回填不工作

Question

我正在使用 Airflow 运行工作流 DAG。我设置的开始日期是 5/26/2014，schedule_interval 是 1 天，没有结束日期，它使用的是 Celery Executor。 Airflow 从给定日期开始运行我的任务，并按预期从开始日期开始每天递增。

但问题是，它只会从开始日期开始安排和运行 16 天，然后停止。我必须重新启动调度程序服务才能安排下一组 dag 并运行。

我修改了下面的属性，在airflow.cfg 中将它们设置为更高，但仍然不起作用，

dag_concurrency = 9999
max_active_runs_per_dag = 9999
parallellism = 9999
celeryd_concurrency = 4

我的要求是让 Airflow 从开始日期到当前日期执行我的任务。看起来它可以运行的 dag 数量有限制，但我无法弄清楚。请建议。

我的 Airflow 版本是 1.7.1.2

Answer 1

不确定我是否理解正确，您是否希望从某个日期到接下来的 16 天运行日常工作？
如果是这样，它与airflow.cfg 没有任何关系，如果您有很多Dag，您发布的设置实际上可以“杀死”您的机器，这是关于将并行运行和调度的任务实例.

我想你需要的是定义一个 start_date 和 end_date。
当您构建将承担这项工作的 DAG 时，您就可以做到这一点。
一种方法是定义 default_args，然后在您的 dag 上使用它们。

default_args = {
     'owner': 'userXX',
     'start_date': datetime(2014, 05, 26),
     'end_date': datetime(2014, 06, 11),
     'email': ['mail1..','mail2...'],
    ...
}

mydag = DAG(
           PARENT_ID, default_args=default_args,
           schedule_interval="@daily",
           user_defined_macros=user_defined_macros
)

_some_Operator(
                task_id='task_idxx',
                .. other settings/jobrun...
                dag=mydag,

            )

希望这可以帮助