如果先前的任务执行需要更多时间，则气流计划将被跳过

Question

我的气流 DAG 中有两个任务。一个触发 API 调用（Http 运算符），另一个使用另一个 API（Http 传感器）不断检查其状态。该 DAG 计划每小时 10 分钟运行一次。但有时一次执行可能需要很长时间才能完成，例如 20 小时。在这种情况下，上一个任务运行时的所有计划都不会执行。

例如，假设我 01:10 的工作需要 10 个小时才能完成。原本应该运行的 02:10、03:10、04:10、... 11:10 等时间表被跳过，只执行 12:10 的时间表。

我正在使用本地执行器。我正在使用下面的脚本运行气流服务器和调度程序。 start_server.sh

export AIRFLOW_HOME=./airflow_home;
export AIRFLOW_GPL_UNIDECODE=yes;
export AIRFLOW_CONN_REST_API=http://localhost:5000;
export AIRFLOW_CONN_MANAGEMENT_API=http://localhost:8001;
airflow initdb;
airflow webserver -p 7200;

start_scheduler.sh

export AIRFLOW_HOME=./airflow_home;
# Connection string for connecting to REST interface server
export AIRFLOW_CONN_REST_API=http://localhost:5000;
export AIRFLOW_CONN_MANAGEMENT_API=http://localhost:8001;
#export AIRFLOW__SMTP__SMTP_PASSWORD=**********;
airflow scheduler;

my_dag_file.py

default_args = {
    'owner': 'airflow',
    'depends_on_past': False,
    'start_date': airflow.utils.dates.days_ago(2),
    'email': admin_email_ids,
    'email_on_failure': False,
    'email_on_retry': False

}

DAG_ID = 'reconciliation_job_pipeline'

MANAGEMENT_RES_API_CONNECTION_CONFIG = 'management_api'
DA_REST_API_CONNECTION_CONFIG = 'rest_api'

recon_schedule = Variable.get('recon_cron_expression',"10 * * * *")
dag = DAG(DAG_ID, max_active_runs=1, default_args=default_args,
          schedule_interval=recon_schedule,
          catchup=False)
dag.doc_md = __doc__

spark_job_end_point = conf['sip_da']['spark_job_end_point']
fetch_index_record_count_config_key = conf['reconciliation'][
    'fetch_index_record_count']


fetch_index_record_count = SparkJobOperator(
    job_id_key='fetch_index_record_count_job',
    config_key=fetch_index_record_count_config_key,
    exec_id_req=False,
    dag=dag,
    http_conn_id=DA_REST_API_CONNECTION_CONFIG,
    task_id='fetch_index_record_count_job',
    data={},
    method='POST',
    endpoint=spark_job_end_point,
    headers={
        "Content-Type": "application/json"}
)

job_endpoint = conf['sip_da']['job_resource_endpoint']

fetch_index_record_count_status_job = JobStatusSensor(
    job_id_key='fetch_index_record_count_job',
    http_conn_id=DA_REST_API_CONNECTION_CONFIG,
    task_id='fetch_index_record_count_status_job',
    endpoint=job_endpoint,
    method='GET',
    request_params={'required': 'status'},
    headers={"Content-Type": "application/json"},
    dag=dag,
    poke_interval=15
)

fetch_index_record_count>>fetch_index_record_count_status_job

SparkJobOperator&JobStatusSensor我的自定义类扩展SimpleHttpOperator& HttpSensor.

如果我设置depends_on_past true它会按预期工作吗？此选项的另一个问题是有时状态检查作业会失败。但下一个时间表应该会触发。我怎样才能实现这种行为？

Answer 1

我认为这里主要讨论的点是你设置的是什么，更多细节可以在这里catchup=False找到。因此，气流调度程序将跳过这些任务执行，您将看到您提到的行为。

这听起来好像如果前一个过程花费的时间比预期的时间长，您就需要执行追赶。你可以尝试改变它catchup=True