pyarrow parquet结合multiprocessing读取HDFS无限挂起排查

• 目标：使用pyarrow与multiprocessing并行读取多个HDFS文件
• 现象：单进程串行调用pq.read_table读取HDFS文件完全正常，换成multiprocessing实现相同逻辑时程序无限挂起，无任何异常抛出
• 已完成的初步排查：
  • 怀疑父子进程运行环境存在差异，核对后确认两者环境变量完全一致
  • 添加print打印调试信息、将进程池大小设置为1，问题仍然稳定复现
• 原实现代码：

import pyarrow.parquet as pq

def read_pq(file):
  table = pq.read_table(file)
  return table

##### 串行执行正常 #####
table = read_pq('hdfs://myns/mydata/000000_0')

###### 多进程执行挂起 #####
import multiprocessing
from multiprocessing import Pool

result_async=[]
with Pool(1) as pool:
  result_async.append( pool.apply_async(pq.read_table, args = ('hdfs://myns/mydata/000000_0',)) )
  results = [r.get() for r in result_async]  # 此处永久挂起，无异常抛出
  print(results)    # 预期返回List[pq.Table]

无限挂起的核心原因是multiprocessing默认的fork启动模式与pyarrow的C++底层运行时、HDFS客户端不兼容，和进程池大小、父子进程环境变量差异无关：

• pyarrow的Parquet读取、HDFS客户端逻辑基于C++实现，底层自带全局IO线程池、锁机制，同时libhdfs客户端本身不支持fork安全：父进程在fork子进程前如果已经导入pyarrow、触发了底层运行时初始化，子进程会直接继承父进程的内存状态，包括处于锁定状态的互斥锁、失效的HDFS连接句柄。
• 子进程继承的锁没有对应的执行线程去释放，后续调用pq.read_table时会永久阻塞在锁等待逻辑上，哪怕进程池大小设为1也会复现问题——问题本质是fork继承的损坏运行时状态，和进程间资源竞争没有关系。

按优先级从高到低排查即可：

1. 验证启动方式问题：在创建进程池前，强制把multiprocessing启动方式改为spawn，如果改完后逻辑正常运行，即可确认是fork继承状态导致的问题。注意启动方式设置必须放在主模块入口、创建Pool之前执行。
2. 给异步任务的get()方法加超时参数，比如r.get(timeout=10)，触发超时后可以通过traceback定位子进程具体阻塞的调用栈位置。
3. 开启pyarrow的DEBUG级别日志，观察HDFS连接、读请求的执行进度，确认卡点：

import pyarrow as pa
pa.set_log_level(1)

4. 不要在父进程提前导入pyarrow相关模块，把导入逻辑移到子进程执行的任务函数内部，排除父进程初始化状态的干扰。

使用spawn启动模式+子进程内初始化pyarrow逻辑，同时添加主模块保护（spawn模式要求必须加if __name__ == '__main__'保护），参考代码：

import multiprocessing
from multiprocessing import Pool

def read_pq(file_path):
    # pyarrow导入放在子进程内部，避免父进程提前初始化底层运行时
    import pyarrow.parquet as pq
    return pq.read_table(file_path)

if __name__ == '__main__':
    # 强制使用spawn启动模式，子进程会全新启动Python解释器，不继承父进程C层状态
    multiprocessing.set_start_method('spawn', force=True)
    task_list = []
    # 进程数可根据实际需求调整
    with Pool(processes=4) as pool:
        hdfs_files = [
            'hdfs://myns/mydata/000000_0',
            # 补充其余待读取的HDFS文件路径
        ]
        for f in hdfs_files:
            task_list.append(pool.apply_async(read_pq, args=(f,)))
        # 可根据文件大小调整超时时间，避免大文件读取误触发超时
        results = [t.get(timeout=300) for t in task_list]
    
    print(results)

额外优化 #

如果默认的libhdfs客户端仍有兼容性问题，可以显式初始化pyarrow内置的HDFS文件系统对象，避免依赖系统级的Hadoop环境配置：

def read_pq(file_path):
    import pyarrow.parquet as pq
    from pyarrow.fs import HadoopFileSystem
    # port=0对应HDFS HA高可用模式，连接配置直接在子进程内初始化
    fs = HadoopFileSystem(host="myns", port=0)
    return pq.read_table(file_path, filesystem=fs)

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