仅运行最新的 Airflow DAG答案

【问题标题】：Run only the latest Airflow DAG仅运行最新的 Airflow DAG
【发布时间】：2018-02-16 14:29:05
【问题描述】：

假设我想使用 Airflow 运行一个非常简单的 ETL DAG：它检查 DB2 中的最后插入时间，如果有的话，它会从 DB1 向 DB2 加载更新的行。

有一些可以理解的要求：

每小时安排一次，前几次运行将持续超过 1 小时
- 例如。第一次运行应该处理一个月的数据，并且会持续 72 小时，
- 所以第二次运行应该处理最后 72 小时，它持续了 7.2 小时，
- 第三道工序 7.2 小时，一小时内完成，
- 从那时起，它每小时运行一次。
当 DAG 运行时，不要启动下一个，而是跳过它。
如果超过触发事件的时间，DAG 没有启动，以后不要启动它。
还有其他的有向无环图，有向无环图应该独立执行。

我发现这些参数和运算符有点混乱，它们之间有什么区别？

depends_on_past
catchup
backfill
LatestOnlyOperator

我应该使用哪个，以及哪个 LocalExecutor？

附言。已经有一个非常相似的thread，但并不累。

【问题讨论】：

标签： airflow airflow-scheduler

【解决方案1】：

DAG max_active_runs = 1 结合 catchup = False 可以解决这个问题。

【讨论】：

【解决方案2】：

这个满足我的要求。 DAG 每分钟运行一次，我的“主要”任务持续 90 秒，所以它应该跳过每秒钟运行一次。我使用ShortCircuitOperator 来检查当前运行是否是目前唯一的运行（在airflow db 的dag_run 表中查询），并使用catchup=False 禁用回填。但是我不能正确使用应该做类似事情的LatestOnlyOperator。

DAG 文件

import os
import sys
from datetime import datetime
import airflow
from airflow import DAG
from airflow.operators.python_operator import PythonOperator, ShortCircuitOperator

import foo
import util

default_args = {
    'owner': 'airflow',
    'depends_on_past': True,
    'start_date': datetime(2018, 2, 13), # or any date in the past
    'email': ['services@mydomain.com'],
    'email_on_failure': True}

dag = DAG(
    'test90_dag',
    default_args=default_args,
    schedule_interval='* * * * *',
    catchup=False)

condition_task = ShortCircuitOperator(
    task_id='skip_check',
    python_callable=util.is_latest_active_dagrun,
    provide_context=True,
    dag=dag)

py_task = PythonOperator(
    task_id="test90_task",
    python_callable=foo.bar,
    provide_context=True,
    dag=dag)

airflow.utils.helpers.chain(condition_task, py_task)

util.py

import logging
from datetime import datetime
from airflow.hooks.postgres_hook import PostgresHook

def get_num_active_dagruns(dag_id, conn_id='airflow_db'):
    # for this you have to set this value in the airflow db
    airflow_db = PostgresHook(postgres_conn_id=conn_id)
    conn = airflow_db.get_conn()
    cursor = conn.cursor()
    sql = "select count(*) from public.dag_run where dag_id = '{dag_id}' and state in ('running', 'queued', 'up_for_retry')".format(dag_id=dag_id)
    cursor.execute(sql)
    num_active_dagruns = cursor.fetchone()[0]
    return num_active_dagruns

def is_latest_active_dagrun(**kwargs):
    num_active_dagruns = get_num_active_dagruns(dag_id=kwargs['dag'].dag_id)
    return (num_active_dagruns == 1)

foo.py

import datetime
import time

def bar(*args, **kwargs):
    t = datetime.datetime.now()
    execution_date = str(kwargs['execution_date'])
    with open("/home/airflow/test.log", "a") as myfile:
        myfile.write(execution_date + ' - ' + str(t) + '\n')
    time.sleep(90)
    with open("/home/airflow/test.log", "a") as myfile:
        myfile.write(execution_date + ' - ' + str(t) + ' +90\n')
    return 'bar: ok'

致谢：此答案基于this blog post。

【讨论】：

【解决方案3】：

DAG max_active_runs = 1 与 catchup = False 相结合，并在 wait_for_downstream=True 的开头添加一个 DUMMY 任务（类似于 START 任务）。从 LatestOnlyOperator 开始 - 如果之前的执行尚未完成，它将有助于避免重新运行任务。或者将“START”任务创建为 LatestOnlyOperator，并确保第一处理层的所有 Taks 部分都连接到它。但请注意 - 根据文档“请注意，如果给定的 DAG_Run 被标记为外部触发，则永远不会跳过下游任务。”

【讨论】：