使用Flask、flask_executor与PyArrow写Parquet时遇RuntimeError求助

问题根源 #

这个错误是因为PyArrow执行to_parquet时，默认会启用并行列转换（内部依赖concurrent.futures.ThreadPoolExecutor），当该操作在flask_executor的后台线程中运行时，若Flask主请求线程已结束、Python解释器资源开始回收，PyArrow内部尝试提交新线程任务就会触发cannot schedule new futures after interpreter shutdown错误。

解决方案 #

1. 禁用PyArrow的并行处理 #

直接关闭PyArrow内部的线程并行逻辑，避免其在后台线程中创建新任务。通过pandas.to_parquet的use_threads参数实现：

def my_background_save_function(args):
    # ... 其他业务代码 ...
    # 禁用PyArrow线程并行，避免内部提交新线程任务
    data.to_parquet(parquet_file, engine='pyarrow', use_threads=False)
    # ... 其他业务代码 ...

该方案简单直接，适合对列转换性能要求不高的场景。

2. 将flask_executor改为进程池模式 #

flask_executor默认使用线程池，线程与主解释器共享生命周期。改用进程池后，后台任务在独立进程中运行，与Flask主进程的生命周期完全隔离：

from flask_executor import Executor

# 初始化时指定使用进程池
executor = Executor(app, executor_type='processpool')

@app.route('/my_endpoint', methods=['POST'])
def my_endpoint():
    # ... 其他业务代码 ...
    future = executor.submit(my_background_function, args)
    return 'Immediate response to the client'

进程池模式能彻底规避解释器资源回收的影响，适合需要保留PyArrow并行性能的场景。

3. 替换为独立异步任务队列（生产环境推荐） #

如果后台任务逻辑复杂、需要持久化或分布式执行，建议改用Celery这类专业异步任务队列，配合Redis/RabbitMQ作为消息中间件：

• 安装依赖：pip install celery redis
• 配置并提交任务：

from celery import Celery

# 初始化Celery，指定消息中间件和结果存储
celery = Celery(__name__, broker='redis://localhost:6379/0', backend='redis://localhost:6379/0')

@celery.task
def my_background_task(args):
    for account in accounts:
        my_background_save_function(account)

@app.route('/my_endpoint', methods=['POST'])
def my_endpoint():
    # ... 其他业务代码 ...
    # 提交任务到Celery队列
    my_background_task.delay(args)
    return 'Immediate response to the client'

启动Celery Worker：celery -A 你的应用模块名 worker --loglevel=info
这种方案完全隔离任务执行与Flask应用的生命周期，是生产环境的长期最优解。

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