Airflow环境下SQLAlchemy通过Pandas拉取MySQL长查询报错连接丢失

1. 补充MySQL客户端读写超时配置 #

你当前仅配置了建连超时connect_timeout，该参数仅作用于TCP连接建立阶段，对查询执行阶段的超时无约束。你需要在connect_args中新增读写超时参数，适配长查询场景：

connect_args={
    "connect_timeout": 1500,
    # 新增以下两个参数，单位为秒，设置为大于查询总耗时即可
    "read_timeout": 1200,
    "write_timeout": 1200,
}

如果使用pymysql原生连接，同理要在连接初始化时传入这两个参数。

2. 配置会话级服务端超时参数 #

MySQL服务端默认的net_read_timeout（等待客户端读取数据的超时时间）通常为30-60秒，wait_timeout（空闲连接超时）默认是8小时，但若DBA做过优化可能会调整的更小。你可以在建立连接后先执行会话级超时设置，避免全局修改服务端配置：

conn = engine.connect()
# 新增以下两行，设置当前会话的读写超时
conn.execute("SET SESSION net_read_timeout = 1200")
conn.execute("SET SESSION net_write_timeout = 1200")
df = pd.read_sql_query(statement, conn)

3. 排查中间网络设备的空闲连接超时 #

你出现的5分钟左右断连的现象，非常符合云环境NAT网关、负载均衡、防火墙等网络设备的默认空闲连接断开阈值（多数云厂商默认是300秒）。本地运行正常是因为本地到MySQL的网络路径没有这类中间设备，或者空闲超时阈值更高。
如果确认是网络设备的问题，有两种解决方式：

• 联系运维调整对应网络设备的空闲超时阈值到1200秒以上
• 使用pandas流式读取结果，避免连接空闲：仅需在read_sql_query中添加chunksize参数，分批拉取结果，让连接持续有数据传输，不会被判定为空闲。该方案不会修改你的SQL计算逻辑，仅调整结果拉取方式：

# 分批拉取后合并为完整DataFrame，chunksize可根据表字段数量调整
dfs = []
for chunk in pd.read_sql_query(statement, conn, chunksize=1000):
    dfs.append(chunk)
df = pd.concat(dfs, ignore_index=True)

4. 排查Airflow侧的配置覆盖 #

如果你使用Airflow的Connections管理MySQL连接，需要确认：

• Airflow连接配置中的Extra字段是否配置了会覆盖超时的参数
• 你代码中实际拿到的connections_string是否和本地测试用的完全一致，可以添加日志打印确认
• Airflow任务是否配置了execution_timeout，若该值小于查询耗时也会导致任务被提前终止，但这种情况报错会明确是Airflow的超时错误，而非MySQL连接断开。

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