cron - Airflow 的 DAG 在一分钟内运行多次,尽管它计划每 5 分钟运行一次

标签 cron airflow workflow scheduler

我创建了一个 DAG,它使用 cron 语法安排在每 5 分钟执行一次。 此外,池是为这个 dag 创建的,只有一个插槽。

我已尝试重新启动服务器/调度程序并重置数据库。目前,DAG 以 UTC 时间运行。另外,我尝试设置我的本地时区,即“欧洲/明斯克”(UTC+3) - 它没有任何效果。

import random
import time
import airflow
from airflow.models import DAG
from airflow.operators.bash_operator import BashOperator
from airflow.operators.python_operator import PythonOperator
from datetime import datetime, timedelta


default_args = {
    'owner': 'airflow',
    'depends_on_past': False,
    'start_date': airflow.utils.dates.days_ago(2),
    'email': ['airflow@example.com'],
    'email_on_failure': False,
    'email_on_retry': False,
    'retries': 1,
    'retry_delay': timedelta(minutes=5),
    'pool': 'download',
    # 'priority_weight': 10,
    # 'queue': 'bash_queue',
}

params = {
    'table': 'api_avitoimage',
}

dag = DAG(
    dag_id='test_download_avitoimage',
    default_args=default_args,
    schedule_interval='*/5 * * * *',
)


def sleep_for_a_bit(random_base):
    time.sleep(random_base)

with dag:

    download = BashOperator(
        task_id='download',
        bash_command='/usr/bin/python3 /home/artur/downloader.py --table {{ params.table }}',
        params=params,
        dag=dag)

    sleep = PythonOperator(
        task_id='sleep_for_a_bit',
        python_callable=sleep_for_a_bit,
        op_kwargs={'random_base': random.uniform(0, 1)},
        dag=dag,
    )

    download >> sleep

问题: DAG 每分钟运行约 2-3 次,这完全是不正确的执行。 已编辑:碰巧有 16/16 个同时活跃的 DAG 运行。但我不明白这个“神奇数字 16”是从哪里来的。

最佳答案

默认情况下,Airflow 会尝试完成自 start_date 以来所有“错过”的 DAG。由于您的 start_date 设置为 airflow.utils.dates.days_ago(2),Airflow 将在按计划开始启动 DAG 之前运行 DAG 576 次。您可以通过将 catchup = False 添加到 DAG 定义(不是 default_args)来关闭它。

魔数(Magic Number)16来自默认设置的参数max_active_runs_per_dag = 16

关于cron - Airflow 的 DAG 在一分钟内运行多次,尽管它计划每 5 分钟运行一次,我们在Stack Overflow上找到一个类似的问题: https://stackoverflow.com/questions/56593421/

相关文章:

linux - 如何使用 curl 调用 php 先获取文件 url 然后下载?

python - 在 Windows 7 上使用 Python CronTab 调度 Python 脚本

linux - 从 crontab 运行时找不到 Sqlplus 命令

c# - 在 ASP.NET 中使用工作流程?

python - 在 python 中设计工作流/流程图表示?

mercurial - 在 hg 中共享单个文件的更改(而不是整个更改集)

cron - 从 Kubernetes Cron 作业调用端点

python - 有没有办法将任务的返回值存储在Python变量中并与下游任务共享(不使用xcom或airflow变量)

airflow - Airflow 传感器中的模式 "reschedule"如何工作?

kubernetes - Apache Airflow 在 AWS 中使用 kubernetes 运行 dag 运算符