动态生成Airflow DAG时Polars DataFrame过滤操作导致任务无限挂起

我之前在Airflow里用Polars的时候也踩过类似的坑，结合你的代码和环境版本（Airflow 2.7.3 + Polars 0.20.31），这个无限挂起的情况大概率是Polars多线程操作和Airflow任务执行环境的资源隔离机制冲突导致的死锁，或者是旧版本Polars在受限环境下的隐藏bug。下面给你拆解可能的原因和实用的解决方案：

可能的触发原因 #

1. 多线程资源冲突：Polars默认会用多线程执行过滤这类操作，而Airflow的执行器（比如Celery、Kubernetes）会给每个任务做资源隔离，这种情况下很容易出现线程死锁，导致操作无限等待。
2. Polars版本bug：你用的0.20.31是比较早的版本了，这个版本在沙箱化的任务环境里，确实存在一些未修复的线程相关问题。
3. 隐藏异常未暴露：Airflow任务里如果发生线程层面的异常，有时候不会直接抛出，反而会让任务看起来像是“挂起”，实际是卡在异常处理的环节。

针对性解决方案 #

方案1：强制Polars用单线程执行

在任务函数开头显式设置Polars使用单线程，避免和Airflow的执行环境抢资源：

def print_hello():
    print("starting")
    # 强制Polars使用单线程，避免多线程冲突
    pl.Config.set_threads(1)
    
    df = pl.DataFrame({
        "key": ["A", "B", "A"],
        "branch": ["br1", "ooo", "br2"],
        "chain": ["ch1", "Y", "ch2"]
    })

    print(df)
    print("before filter") 
    chains = df.filter(pl.col("key") == "A").select("chain").to_series().to_list()
    print("after filter")
    print(chains)
    print("finish dag")

方案2：升级Polars到最新稳定版

旧版本的Polars在隔离环境下的线程问题已经在后续版本里修复了不少，直接升级到最新稳定版大概率能解决问题：

pip install --upgrade polars

升级后可以先在本地跑一遍print_hello函数，确认没问题再部署到Airflow上。

方案3：加异常捕获排查隐藏问题

在任务函数里套一层try-except，把所有异常都打出来，避免因为隐藏异常导致的“假挂起”：

def print_hello():
    import traceback
    try:
        print("starting")

        df = pl.DataFrame({
            "key": ["A", "B", "A"],
            "branch": ["br1", "ooo", "br2"],
            "chain": ["ch1", "Y", "ch2"]
        })

        print(df)
        print("before filter") 
        chains = df.filter(pl.col("key") == "A").select("chain").to_series().to_list()
        print("after filter")
        print(chains)
        print("finish dag")
    except Exception as e:
        print(f"任务执行出错：{str(e)}")
        traceback.print_exc()

这样即使有异常，也能在Airflow的任务日志里看到完整的错误栈，方便精准定位问题。

方案4：改用Lazy API执行操作

试试用Polars的延迟执行API来做过滤，可能绕过eager模式下的线程问题：

chains = df.lazy()\
           .filter(pl.col("key") == "A")\
           .select("chain")\
           .collect()\
           .to_series()\
           .to_list()

额外测试建议 #

• 先在本地直接跑print_hello函数，确认逻辑本身没问题（你应该已经试过了）；
• 临时把Airflow的执行器换成SequentialExecutor（适合测试用），看看任务能不能正常跑，排除分布式执行器的隔离问题。

备注：内容来源于stack exchange，提问作者elvainch