运行中的气流 dag 问题 - 运行循环调度

Question

我按计划在远程 ssh 上为运行命令创建了这个 dag。

from datetime import timedelta
from airflow import DAG
from airflow.operators.bash_operator import BashOperator
from airflow.utils.dates import days_ago
from airflow.contrib.hooks.ssh_hook import SSHHook as sscon
from airflow.contrib.operators.ssh_operator import SSHOperator


default_args = {
    'owner': 'airflow',
    'depends_on_past': False,
    'start_date': days_ago(0),
    'email': ['e@example.com'],
    'email_on_failure': False,
    'email_on_retry': False,
    #'retries': 1,
    #'retry_delay': timedelta(minutes=10),
}
dag = DAG(
    'ssh_second',
    default_args=default_args,
    description='A simple bash DAG LAB',
    schedule_interval=timedelta(minutes=1),
    tags=['test'],
)

sshcon = sscon(remote_host="192.168.1.250", username="user", password="password", port=22)

t1 = BashOperator(
    task_id='echo1',
    bash_command='echo "simple task! by dag" ',
    dag=dag,
)

t5 = SSHOperator(
    task_id="remote-connection",
    command="/bin/date >> /home/user/date.txt && echo 'from airflow' >> /home/user/date.txt",
    ssh_hook=sshcon,
    dag=dag)

t1 >> t5

它工作正常，但气流运行它每秒都在流动，而且势不可挡。

tree view of loop run

不知道问题出在哪里！

Answer 1

您的 schedule_interval 每分钟设置一次：schedule_interval=timedelta(minutes=1) 这就是导致 DAG 如此频繁执行的原因。将 schedule_interval 更新为频率较低的间隔。

此外，我 强烈建议 将 start_date 修改为静态值而不是动态值（例如 days_ago(0)）。这个 guide 和其他的在我开始使用 Airflow 时非常有用。设置静态 start_date 后，如果您不希望 DAG 执行从 start_date 到现在的“缺失”运行，则可以设置 catchup=False。

运行中的气流 dag 问题 - 运行循环调度

airflow dag problem in run - run schedule in loop

airflow

schedule

airflow-scheduler

运行 中的气流 dag 问题 - 运行 循环调度

airflow dag problem in run - run schedule in loop

airflow

schedule

airflow-scheduler

运行中的气流 dag 问题 - 运行循环调度