Airflow DAG未执行求助：调度器端口占用+任务状态异常

问题概述 #

本地部署Airflow后，手动触发DAG出现以下异常：

• 任务状态时而为queued，时而显示成功但运行时长为00:00:00（实际未执行）
• WebUI Graph视图始终显示「Dag Has Yet To Run」
• 启动scheduler时出现端口8793被占用的报错

原始DAG代码 #

from datetime import datetime
from airflow import DAG
from airflow.models import Variable
from airflow.operators.python import PythonOperator

def get_var():
    #a=Variable.get('abc')
    print('abd')

with DAG(dag_id='test_var',start_date=datetime.now()) as dag:
    task1=PythonOperator(task_id='var',python_callable=get_var)

Scheduler启动报错日志 #

[2022-10-31 09:46:45,562] {scheduler_job.py:701} INFO - Starting the scheduler
[2022-10-31 09:46:45,562] {scheduler_job.py:706} INFO - Processing each file at most -1 times
[2022-10-31 09:46:45,565] {executor_loader.py:107} INFO - Loaded executor: SequentialExecutor
[2022-10-31 09:46:45,569] {manager.py:163} INFO - Launched DagFileProcessorManager with pid: 13315
[2022-10-31 09:46:45,570] {scheduler_job.py:1381} INFO - Resetting orphaned tasks for active dag runs
[2022-10-31 09:46:46,169] {settings.py:58} INFO - Configured default timezone Timezone('UTC')
[2022-10-31T09:46:46.172+0800] {manager.py:409} WARNING - Because we cannot use more than 1 thread (parsing_processes = 2) when using sqlite. So we set parallelism to 1.
[2022-10-31 09:46:46 +0800] [13314] [INFO] Starting gunicorn 20.1.0
[2022-10-31 09:46:46 +0800] [13314] [ERROR] Connection in use: ('::', 8793)
[2022-10-31 09:46:46 +0800] [13314] [ERROR] Retrying in 1 second.
[2022-10-31 09:46:47 +0800] [13314] [ERROR] Connection in use: ('::', 8793)
[2022-10-31 09:46:47 +0800] [13314] [ERROR] Retrying in 1 second.
[2022-10-31 09:46:48 +0800] [13314] [ERROR] Connection in use: ('::', 8793)
[2022-10-31 09:46:48 +0800] [13314] [ERROR] Retrying in 1 second.
[2022-10-31 09:46:49 +0800] [13314] [ERROR] Connection in use: ('::', 8793)
[2022-10-31 09:46:49 +0800] [13314] [ERROR] Retrying in 1 second.
[2022-10-31 09:46:50 +0800] [13314] [ERROR] Connection in use: ('::', 8793)
[2022-10-31 09:46:50 +0800] [13314] [ERROR] Retrying in 1 second.
[2022-10-31 09:46:51 +0800] [13314] [ERROR] Can't connect to ('::', 8793)

解决方案 #

1. 修复start_date的错误用法 #

核心问题：start_date=datetime.now()会导致每次DAG文件被解析时，start_date都被设置为当前时间，Airflow会认为这是一个新的DAG版本，无法生成有效的调度记录，手动触发也会出现异常。

修改方法：使用固定的过去时间，或者Airflow提供的days_ago工具函数：

• 方法一：用固定日期

from datetime import datetime
from airflow import DAG
from airflow.models import Variable
from airflow.operators.python import PythonOperator

def get_var():
    #a=Variable.get('abc')
    print('abd')

# 使用固定的过去日期
with DAG(dag_id='test_var', start_date=datetime(2022, 10, 30)) as dag:
    task1=PythonOperator(task_id='var',python_callable=get_var)

• 方法二：用days_ago（推荐）

from datetime import datetime
from airflow import DAG
from airflow.models import Variable
from airflow.operators.python import PythonOperator
from airflow.utils.dates import days_ago  # 导入工具函数

def get_var():
    #a=Variable.get('abc')
    print('abd')

# 设置为1天前的时间
with DAG(dag_id='test_var', start_date=days_ago(1)) as dag:
    task1=PythonOperator(task_id='var',python_callable=get_var)

2. 解决Scheduler端口8793被占用问题 #

方式一：杀掉占用端口的进程

• Windows系统：
  1. 查找占用端口的进程ID：netstat -ano | findstr :8793
  2. 强制杀掉进程：taskkill /PID <进程ID> /F
• Linux/macOS系统：
  1. 查找占用端口的进程ID：lsof -i :8793 或 netstat -tulpn | grep 8793
  2. 强制杀掉进程：kill -9 <进程ID>

方式二：修改Airflow配置更换端口

1. 找到Airflow的配置文件airflow.cfg（默认路径：~/.airflow/airflow.cfg）
2. 找到scheduler_port配置项，将默认的8793修改为其他未被占用的端口（比如8794）
3. 保存配置后，重新启动scheduler

3. 修正Airflow初始化顺序 #

正确的初始化流程应该是：

1. 初始化数据库：airflow db init
2. 启动Webserver：airflow webserver -p 8080
3. 启动Scheduler：airflow scheduler

之前先启动Webserver再初始化数据库的顺序，可能导致数据库状态不一致，影响DAG的正常调度。

内容的提问来源于stack exchange，提问作者lynn kuo