仅运行最新的Airflow DAG

问题描述 投票:0回答:2

假设我想用Airflow运行一个非常简单的ETL DAG:它检查DB2中的最后一次插入时间,并且它将DB1中的新行加载到DB2(如果有的话)。

有一些可以理解的要求:

  1. 它按小时计划,前几次运行将持续超过1小时 例如。第一次运行应该处理一个月的数据,并持续72小时, 所以第二次运行应该处理最后72小时,它持续7.2小时, 第三个过程7.2小时,一小时内完成, 从那时起它每小时运行一次。
  2. 当DAG正在运行时,不要启动下一个,而是跳过它。
  3. 如果时间超过了触发事件,并且DAG没有启动,请不要随后启动它。
  4. 还有其他DAG,DAG应该独立执行。

我发现这些参数和操作符有点混乱,它们之间的区别是什么?

  • depends_on_past
  • catchup
  • backfill
  • LatestOnlyOperator

我应该使用哪一个,以及哪个LocalExecutor?

PS。已经有一个非常相似的thread,但它并不疲惫。

airflow airflow-scheduler
2个回答
4
投票

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


0
投票

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

DAY文件

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': ['[email protected]'],
    '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)

UT IL.朋友

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.朋友

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

© www.soinside.com 2019 - 2024. All rights reserved.