Airflow 2.5.0对接MSSQL 19后端遇死锁及调度异常求助

环境信息 #

• Airflow版本：2.5.0
• 后端数据库：MSSQL 19

问题现象 #

1. 运行含时区配置的DAG时，MSSQL触发事务死锁（错误码1205），直接导致Airflow调度器崩溃，死锁发生在UPDATE dag_run语句执行阶段
2. 移除DAG的时区配置后，DAG无法自动触发调度，仅支持手动触发执行

相关DAG代码 #

带时区配置的DAG（my_tz_dag） #

import pendulum
from airflow import DAG
from airflow.operators.python_operator import PythonOperator
from datetime import timedelta

# 定义带时区的DAG
dag = DAG("my_tz_dag", start_date=pendulum.datetime(2016, 1, 1, tz="Europe/Amsterdam"), schedule_interval=timedelta(minutes=15))

# 任务函数
def print_hello():
    print("Hello, world!")

# 定义Python任务
task = PythonOperator(task_id="print_hello", python_callable=print_hello, dag=dag)

无时区配置的定时清理DAG（log_cleaning） #

from datetime import datetime, timedelta
from airflow import DAG
from airflow.operators.bash_operator import BashOperator

default_args = {
    'owner': 'airflow',
    'depends_on_past': False,
    'start_date': datetime(2023, 3, 1),
    'retries': 1,
    'retry_delay': timedelta(minutes=5)
}

dag = DAG(
    'log_cleaning',
    default_args=default_args,
    schedule_interval='*/5 * * * *',
    catchup=False
)
log_cleaning = BashOperator(
    task_id='log_cleaning',
    bash_command='find /usr/local/airflow/logs/* -mtime +7 -exec rm {} \;',
    dag=dag
)

调度器崩溃日志 #

[Microsoft][ODBC Driver 17 for SQL Server][SQL Server]事务（进程ID 194）与另一个进程在锁资源上发生死锁，已被选为死锁牺牲品。请重新运行该事务。(1205) (SQLExecDirectW)
[SQL语句: UPDATE dag_run SET last_scheduling_decision=?, updated_at=? WHERE dag_run.id = ?]
[参数: ((datetime.datetime(2023, 4, 3, 18, 43, 12, 737104, tzinfo=Timezone('UTC')), ...)]

解决方案 #

针对时区导致的死锁问题 #

1. 对齐时区配置：确保MSSQL服务器时区与Airflow配置文件airflow.cfg中的default_timezone一致，避免时区转换引发的锁竞争
2. 调优调度器参数：
   • 增大scheduler_parsing_processes值（默认2），分散单个进程的调度压力
   • 开启scheduler_use_task_family，优化任务调度时的数据库查询逻辑
3. 统一DAG时区：业务层面若需本地时间显示，可在任务内部做转换；DAG层面统一使用UTC时区，规避跨时区转换带来的数据库锁冲突
4. 排查死锁根源：用SQL Server Profiler或Extended Events捕获死锁图，明确死锁涉及的进程与资源，针对性调整索引或事务隔离级别

针对无时区DAG无法自动调度的问题 #

1. 检查Airflow核心配置：确保airflow.cfg中default_timezone设为UTC（Airflow推荐配置），load_examples设为False避免示例DAG干扰
2. 验证DAG基础参数：确认start_date为过去的时间点，schedule_interval格式正确（如*/5 * * * *代表每5分钟执行一次）
3. 重启并检查调度器：通过airflow scheduler logs查看是否有其他报错，执行以下命令重启调度器：

   airflow scheduler restart
   

4. 确认DAG启用状态：在Airflow UI中检查DAG开关是否开启，catchup设置符合预期（catchup=False表示不补跑历史任务）

内容的提问来源于stack exchange，提问作者rahul ranjan