Airflow多连接运行ETL报utf-16le解码错误的集成方案咨询

三个报错是连锁触发，不是独立问题：

• 核心诱因是Docker Ubuntu环境下pyodbc依赖的ODBC Driver/FreeTDS默认配置和Windows原生SQL Server驱动逻辑不一致：10并发场景下，部分TCP连接被Docker网络、防火墙或数据库侧主动断开时，驱动会拿到截断的半包返回数据，直接触发utf-16le解码失败；如果异常阻塞时间超过Airflow worker默认的杀进程阈值，就会收到SIGTERM信号；通信链路被异常断开时pyodbc就会抛出08S01错误。
• Windows PyCharm环境运行正常是因为Windows自带的SQL Server Native Client内置了断连重试、坏包自动截断兜底逻辑，不会把不完整的编码字节直接抛给上层应用。
• 手动重启任务随机成功/失败，也符合偶发坏连接、网络闪断的特征，不是代码逻辑的硬错误。

1. 封装自定义解码逻辑 #

先把utf-16le兜底解码、转换器注册的逻辑抽成独立公共函数，不要散落在业务代码里：

import pyodbc
from typing import Optional

def _safe_utf16le_decode(raw_bytes: Optional[bytes]) -> str:
    """兜底处理SQL Server返回的截断、非法utf-16le字节"""
    if not raw_bytes:
        return ""
    try:
        return raw_bytes.decode("utf-16le")
    except UnicodeDecodeError:
        # 逐次截断末尾1-3字节（utf-16le单字符占2字节，最多截2个半字符即可覆盖）重试
        for trim in range(1, 4):
            try:
                return raw_bytes[:-trim].decode("utf-16le", errors="ignore")
            except UnicodeDecodeError:
                continue
        # 极端场景替换非法字符返回，不中断任务
        return raw_bytes.decode("utf-16le", errors="replace")

def _init_conn_converters() -> None:
    """给当前pyodbc连接注册自定义解码钩子，覆盖所有宽字符串类型"""
    # 三个类型ID分别对应NVARCHAR、NCHAR、NTEXT
    pyodbc.add_output_converter(pyodbc.SQL_WVARCHAR, _safe_utf16le_decode)
    pyodbc.add_output_converter(pyodbc.SQL_WCHAR, _safe_utf16le_decode)
    pyodbc.add_output_converter(pyodbc.SQL_WLONGVARCHAR, _safe_utf16le_decode)

注意：pyodbc的output_converter是连接级别配置，全局注册只对注册后新建的连接生效，必须每次新建连接后重新调用注册逻辑，不要只在DAG文件顶层全局注册一次。

2. 修改两个数据库连接构造函数 #

在dwh_conn和gladiator_conn的构造逻辑里，连接创建成功后立刻调用转换器注册函数，同时加基础连接配置减少断连：

def get_dwh_conn() -> pyodbc.Connection:
    # 替换成你原来的连接串逻辑，建议加上超时参数
    conn = pyodbc.connect(
        "DRIVER={ODBC Driver 17 for SQL Server};SERVER=你的DWH地址;DATABASE=库名;UID=账号;PWD=密码;Connection Timeout=30;",
        timeout=30
    )
    # 注册自定义解码器
    _init_conn_converters()
    # 关闭自动提交按需调整，加会话配置减少驱动兼容性问题
    conn.autocommit = False
    conn.execute("SET ARITHABORT ON;SET ANSI_WARNINGS ON;")
    return conn

def get_gladiator_conn() -> pyodbc.Connection:
    conn = pyodbc.connect(
        "DRIVER={ODBC Driver 17 for SQL Server};SERVER=你的Gladiator地址;DATABASE=库名;UID=账号;PWD=密码;Connection Timeout=30;",
        timeout=30
    )
    _init_conn_converters()
    conn.autocommit = False
    conn.execute("SET ARITHABORT ON;SET ANSI_WARNINGS ON;")
    return conn

3. 改造ETL任务函数，加重试和连接生命周期管理 #

不要在任务里复用全局长连接，每个任务实例运行时新建连接、用完即关，同时对数据库操作加异常重试，捕获通信故障、解码错误时自动重建连接重试，从业务层兜底08S01错误：

from tenacity import retry, stop_after_attempt, wait_exponential, retry_if_exception_type

# 重试配置：最多重试3次，指数退避等待，只捕获数据库通信、解码类异常
@retry(
    stop=stop_after_attempt(3),
    wait=wait_exponential(multiplier=1, min=2, max=10),
    retry=retry_if_exception_type((pyodbc.OperationalError, UnicodeDecodeError))
)
def _query_with_retry(cursor, query: str, params: tuple = ()):
    cursor.execute(query, params)
    if cursor.description:
        cols = [col[0] for col in cursor.description]
        return [dict(zip(cols, row)) for row in cursor.fetchall()]
    return cursor.rowcount

def update_from_gladiator_ost(**context):
    # 每次任务运行新建连接，禁止复用全局连接对象
    g_conn = None
    dwh_conn = None
    try:
        g_conn = get_gladiator_conn()
        dwh_conn = get_dwh_conn()
        g_cursor = g_conn.cursor()
        dwh_cursor = dwh_conn.cursor()

        # 替换成你原来的抽取逻辑，所有SQL执行走_query_with_retry
        source_data = _query_with_retry(
            g_cursor,
            "SELECT * FROM 源表 WHERE update_time >= ?",
            (context["data_interval_start"],)
        )

        # 替换成你原来的转换、写入DWH逻辑，写入操作也走带重试的封装
        # ... 你的原有业务逻辑 ...

        dwh_conn.commit()
    except Exception as e:
        if dwh_conn:
            dwh_conn.rollback()
        raise e
    finally:
        # 无论成功失败都关闭连接，不残留死连接
        if g_conn:
            g_conn.close()
        if dwh_conn:
            dwh_conn.close()

对应Airflow任务定义里，记得给任务设置合理的execution_timeout，比如设成1800秒，避免长查询被worker默认的短超时误发SIGTERM杀掉。

4. Docker环境配置优化（根源减少异常） #

光改代码只能兜底，还要调整部署配置从根源减少断连：

• 在Airflow镜像的ODBC配置文件/etc/odbc.ini里，给SQL Server驱动加TCP Keepalive配置：KeepAlive=30、KeepAliveInterval=10，避免空闲连接被Docker网桥、防火墙踢掉
• 调整Airflow worker配置：把worker_kill_seconds从默认60调到300，避免长查询被误杀
• 控制单worker的并发数：总并发10的场景下，不要让单个worker同时持有超过5个数据库连接，避免触发SQL Server端的连接阈值主动断连

• 改完后先压测30分钟，开15并发跑简单查询任务，确认没有08S01、解码错误后再上正式业务逻辑
• 如果仍有偶发解码错误，可以在_safe_utf16le_decode里加日志打印异常字节的长度和前缀，排查是不是源库特定字段本身存在脏数据
• 不要用全局连接对象跨任务复用，Airflow多进程/多worker运行时全局连接会直接失效触发通信错误

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