45

我刚刚开始使用 Airbnb 的气流，我仍然不清楚如何/何时完成回填。

具体来说，有两个用例让我感到困惑：

  1. 如果我运行airflow scheduler几分钟，停止一分钟，然后再次重新启动，我的 DAG 似乎在前 30 秒左右运行了额外的任务，然后它继续正常运行（每 10 秒运行一次）。这些额外的任务是否“回填”了在早期运行中无法完成的任务？如果是这样，我将如何告诉气流不要回填这些任务？

  2. 如果我运行airflow scheduler几分钟，然后运行airflow clear MY_tutorial，然后重新启动airflow scheduler，它似乎运行了大量的额外任务。这些任务是否也以某种方式“回填”任务？或者我错过了什么。

目前，我有一个非常简单的 dag：

default_args = {
    'owner': 'me',
    'depends_on_past': False,
    'start_date': datetime(2016, 10, 4),
    'email': ['airflow@airflow.com'],
    'email_on_failure': False,
    'email_on_retry': False,
    'retries': 1,
    'retry_delay': timedelta(minutes=5),
    # 'queue': 'bash_queue',
    # 'pool': 'backfill',
    # 'priority_weight': 10,
    # 'end_date': datetime(2016, 1, 1),
}

dag = DAG(
    'MY_tutorial', default_args=default_args, schedule_interval=timedelta(seconds=10))

# t1, t2 and t3 are examples of tasks created by instantiating operators
t1 = BashOperator(
    task_id='print_date',
    bash_command='date',
    dag=dag)

t2 = BashOperator(
    task_id='sleep',
    bash_command='sleep 5',
    retries=3,
    dag=dag)

templated_command = """
    {% for i in range(5) %}
        echo "{{ ds }}"
        echo "{{ macros.ds_add(ds, 8)}}"
        echo "{{ params.my_param }}"
    {% endfor %}
"""

t3 = BashOperator(
    task_id='templated',
    bash_command=templated_command,
    params={'my_param': 'Parameter I passed in'},
    dag=dag)

second_template = """
    touch ~/airflow/logs/test
    echo $(date) >> ~/airflow/logs/test
"""

t4 = BashOperator(
    task_id='write_test',
    bash_command=second_template,
    dag=dag)

t1.set_upstream(t4)
t2.set_upstream(t1)
t3.set_upstream(t1)

我在气流配置中更改的唯一两件事是

  1. 我从使用 sqlite db 更改为使用 postgres db
  2. 我正在使用 aCeleryExecutor而不是 aSequentialExecutor

非常感谢你的帮助！

4

3 回答 3

71

当您将 DAG 的调度程序切换更改为“on”时，调度程序将触发所有未记录状态的 dag 运行实例的回填，从您在“default_args”中指定的 start_date 开始。

例如：如果开始日期是“2017-01-21”，并且您在“2017-01-22T00:00:00”打开了调度开关，并且您的 dag 配置为每小时运行，那么调度程序将回填 24 dag运行，然后按计划的时间间隔开始运行。

这基本上是您的两个问题中正在发生的事情。在 #1 中，它正在填充您关闭调度程序的 30 秒内缺少的 3 次运行。在 #2 中，它填充了从 start_date 到“现在”的所有 DAG 运行。

有两种方法可以解决这个问题：

  1. 将 start_date 设置为将来的某个日期，以便它仅在到达该日期后才开始调度 dag 运行。请注意，如果您更改 DAG 的 start_date，则由于开始日期存储在气流数据库中的方式，您还必须更改 DAG 的名称。
  2. 使用“-m”（--mark-success）标志从命令行手动运行回填，该标志告诉气流不要实际运行 DAG，而只是在数据库中将其标记为成功。

例如

airflow backfill MY_tutorial -m -s 2016-10-04 -e 2017-01-22T14:28:30
于 2017-01-22T12:32:27.703 回答
19

请注意，从 1.8 版开始，Airflow 允许您使用追赶来控制此行为。catchup_by_default=False在airflow.cfg 或 catchup=False您的DAG 定义中设置。

请参阅https://airflow.apache.org/scheduler.html#backfill-and-catchup

于 2018-08-08T17:57:47.250 回答
-1

Airflow 的 UI 上的 On/Off 仅显示“暂停”，这意味着，如果它为 ON，它只会在触发时暂停，如果关闭，它将在该日期再次继续。

于 2018-03-20T07:53:35.360 回答