ProcessPoolExecutor并行拉取DB数据时pd.read_sql执行中断如何解决？

• 首先补全全链路异常捕获，定位具体报错原因
  你当前的代码没有任何异常捕获逻辑，子线程/子进程运行时抛出的异常会直接导致执行中断，且默认不会输出完整错误栈，是最常见的静默退出原因。按照如下方式改造代码即可拿到具体错误信息：

  import traceback
  import pandas as pd
  
  def fetch_report(_date, connection):
      try:
          print(f"Selecting Data for {_date}", flush=True)
          sql = f"""
          -- 你的SQL逻辑
          """
          df = pd.read_sql(sql=sql, con=connection)
          print(df.shape, flush=True)
          return df
      except Exception as e:
          print(f"拉取数据出错 date={_date}, 错误: {str(e)}", flush=True)
          print(traceback.format_exc(), flush=True)
          raise
  
  def task(date_list, _report):
      try:
          # 注意：必须在task内部独立创建数据库连接，禁止复用主进程/其他线程的连接
          db_con = create_your_db_connection() # 替换为你的数据库连接创建逻辑
          df = pd.DataFrame()
          for date in date_list:
              log.info(f"Fetching Data for XYZ: {date}")
              df_tmp = fetch_report(_date=date, connection=db_con)
              # df.append已在pandas 2.0+废弃，替换为concat避免兼容报错
              df = pd.concat([df, df_tmp], ignore_index=True)
          return df
      except Exception as e:
          print(f"任务执行出错 report={_report}, 错误: {str(e)}", flush=True)
          print(traceback.format_exc(), flush=True)
          raise
  

  注意所有print添加flush=True参数，避免多进程场景下输出缓冲导致的日志不显示问题。

• 排查数据库连接的并发安全问题
  绝大多数数据库驱动的连接对象都不是线程/进程安全的，绝对不能在多个执行单元中共享同一个连接：

  1. 禁止在主进程预先创建连接再传入task中使用，多进程场景下fork出来的子进程会复用父进程的TCP连接副本，直接导致连接状态异常中断
  2. 每个task必须独立创建、关闭自己的连接，连接创建逻辑完全放在task函数内部
  3. 如果你使用SQLAlchemy等ORM框架，需要使用线程安全的会话生成方式（如scoped_session），每个线程独立获取会话

• 排查数据库端的并发限制
  数据库通常会有两层并发连接限制，超过限制会直接拒绝连接导致任务中断：

  1. 数据库全局的最大连接数配置（如MySQL的max_connections、PostgreSQL的max_connections）
  2. 你使用的数据库账号的专属连接数限制（如MySQL的max_user_connections、PostgreSQL的ROLE级CONNECTION LIMIT）
     确保你设置的max_workers数值+当前已存在的数据库连接数，不超过上述两个限制的最小值。

• 低并发验证定位问题
  先将ProcessPoolExecutor的max_workers改为1测试：

  1. 如果单worker可以正常运行，说明问题属于并发连接限制/连接共享问题，按照前两步排查即可
  2. 如果单worker依然无法运行，说明是驱动在子进程中的兼容性问题，比如部分数据库驱动需要在子进程中重新初始化底层依赖库，将驱动的初始化逻辑也移动到task函数内部即可。

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