在 Airflow 版本 2.2.5 中进行 dag 解析时面临任务超时错误答案

【问题标题】：Facing task Timeout Error while dag parsing in Airflow version 2.2.5在 Airflow 版本 2.2.5 中进行 dag 解析时面临任务超时错误
【发布时间】：2022-06-14 14:39:20
【问题描述】：

我遇到了Airflow Version 2.2.5/Composer 2.0.15 的任务超时错误。相同的代码在Airflow version2.2.3 /Composer Version 1.18.0 中运行得非常好

错误信息：

Broken DAG: [/home/airflow/gcs/dags/test_dag.py] Traceback (most recent call last):
  File "/opt/python3.8/lib/python3.8/enum.py", line 256, in __new__
    if canonical_member._value_ == enum_member._value_:
  File "/opt/python3.8/lib/python3.8/site-packages/airflow/utils/timeout.py", line 37, in handle_timeout
    raise AirflowTaskTimeout(self.error_message)
airflow.exceptions.AirflowTaskTimeout: DagBag import timeout for /home/airflow/gcs/dags/test_dag.py after 30.0s.
Please take a look at these docs to improve your DAG import time:
* https://airflow.apache.org/docs/apache-airflow/2.2.5/best-practices.html#top-level-python-code
* https://airflow.apache.org/docs/apache-airflow/2.2.5/best-practices.html#reducing-dag-complexity, PID: 1827

根据文档或有关顶级 Python 代码的错误消息中的链接。我们为 Dags 和任务制定了框架。

主文件夹

|___ 天

|___ 任务

|___ 库

a) 所有主要的 dag 文件都在 dags 文件夹中

b) 实际的函数或查询（来自 PythonOperator 函数/Sql Queries）放在 tasks 文件夹下的 *.py 文件中

c) 常用功能放在 libs 文件夹下的 python 文件中。

在这里提供基本的 dag 结构：

# Import libraries and functions 
import datetime

from airflow import models, DAG
from airflow.contrib.operators import bigquery_operator, bigquery_to_gcs, bigquery_table_delete_operator
from airflow.operators.python_operator import PythonOperator
from airflow.operators.bash_operator import BashOperator
##from airflow.executors.sequential_executor import SequentialExecutor
from airflow.utils.task_group import TaskGroup

## Import codes from tasks and libs folder
from libs.compres_suppress.cot_suppress import *
from libs.teams_plugin.teams_plugin import *
from tasks.email_code.trigger_email import *

# Set up Airflow DAG
default_args = {
    'owner': 'airflow',
    'depends_on_past': False,
    'start_date': datetime.datetime(2020, 12, 15, 0),
    'retries': 1,
    'retry_delay': datetime.timedelta(minutes=1),
    'on_failure_callback': trigger_email
}

DAG_ID = 'test_dag'


# Check exscution date
if "<some condition>" matches:
    run_date = <date in config file>
else:
    run_date = datetime.datetime.now().strftime("%Y-%m-%d")
    run_date_day = datetime.datetime.now().isoweekday()


dag = DAG(
    DAG_ID,
    default_args=default_args, catchup=False,
    max_active_runs=1, schedule_interval=SCHEDULE_INTERVAL
)


next_dag_name = "next_dag1"
if env == "prod":
    if run_date_day == 7:
        next_dag_name = "next_dag2"
    else:
        next_dag_name = "next_dag1"


run_id = datetime.datetime.now().strftime("%Y-%m-%dT%H:%M:%S")

# Define Airflow DAG
with dag:

    team_notify_task = MSTeamsWebhookOperator(
        task_id='teams_notifi_start_task',
        http_conn_id='http_conn_id',
        message=f"DAG has started <br />"
                f"<strong> DAG ID:</strong> {DAG_ID}.<br />",
        theme_color="00FF00",
        button_text="My button",
        dag=dag)
        
    task1_bq = bigquery_operator.BigQueryOperator(
        task_id='task1',
        sql=task1_query(
            table1="table1",
            start_date=start_date),
        use_legacy_sql=False,
        destination_dataset_table="destination_tbl_name",
        write_disposition='WRITE_TRUNCATE'      
    )

    ##### Base Skeletons #####
    with TaskGroup("taskgroup_lbl", tooltip="taskgroup_sample") as task_grp:
         tg_process(args=default_args,run_date=run_date)
         

    if run_mode == "<env_name>" and next_dag != "":
        next_dag_trigg = BashOperator(
            task_id=f'trigger_{next_dag_name}',
            bash_command="gcloud composer environments run " + <env> + "-cust_comp --location us-east1 dags trigger -- " + next_dag_name + " --run-id='trigger_ "'"
        )
        task_grp >> next_dag_trigger
        
    team_notify_task >> task1_bq >> task_grp 
    enter code here

有人可以帮助解决导致问题的原因吗？

【问题讨论】：

标签： airflow google-cloud-composer airflow-2.x

【解决方案1】：

增加 dag/task 超时时间可以解决问题。

转到 Airflow (Web UI)，在顶部栏导航到

变量--> 配置 --> [core] --> dagbag_import_timeout = 。

如果使用 Composer，同样可以通过以下步骤完成。

a) 进入 Composer 服务并选择要修改设置的 Composer。

b) 点击AIRFLOW CONFIGURATION OVERRIDES --> EDIT --> (add/edit) dagbag_import_timeout=160

c) 点击保存

【讨论】：