Pandas read_parquet()多进程读取GCS路径时挂起问题求助

• GCS客户端的进程不安全特性：Pandas读取GCS路径依赖gcsfs（或fsspec）底层调用google-cloud-storage客户端，该客户端的默认实例不支持多进程场景。当通过ProcessPoolExecutor创建子进程时，子进程会继承父进程的客户端连接池、HTTP会话状态，这些共享状态在并发访问时会引发锁竞争、连接冲突，最终导致进程挂起死锁。
• Pandas的GCS访问逻辑未做进程隔离：默认情况下，pd.read_parquet(GS_PATH)会复用全局的GCS文件系统实例，多进程环境下多个子进程共享该实例的状态，进一步加剧了资源竞争问题。
• 任务提交逻辑过载：原代码中一次性提交1000个任务，瞬间产生大量GCS请求，可能触发GCS的限流机制或客户端内部的资源耗尽，间接导致挂起。

方案1：子进程内独立初始化GCS文件系统 #

在每个子进程的任务函数中，手动创建独立的GCSFileSystem实例，避免继承父进程的共享状态，将其传入read_parquet的storage_options参数：

import pandas as pd
from gcsfs import GCSFileSystem
from concurrent.futures import ProcessPoolExecutor

def read_chunk(*args):
    # 子进程内独立创建GCS文件系统实例
    gcs = GCSFileSystem()
    df = pd.read_parquet(GS_PATH, storage_options={"fs": gcs})
    # 后续处理逻辑

num_files = 1000
batch_size = 2  # 控制每批提交的任务数，避免过载
with ProcessPoolExecutor(max_workers=2) as executor:
    futures = []
    while num_files > 0:
        current_batch = min(batch_size, num_files)
        for _ in range(current_batch):
            future = executor.submit(read_chunk, *args)
            futures.append(future)
        # 等待当前批任务完成再提交下一批
        for future in futures:
            future.result()
        futures.clear()
        num_files -= current_batch

方案2：配置多进程安全的GCS客户端参数 #

通过storage_options传递客户端配置，禁用共享连接池，强制每个请求使用独立连接：

def read_chunk(*args):
    storage_options = {
        "client_kwargs": {
            "pool_connections": 1,
            "pool_maxsize": 1,
            "disable_retries": False  # 可根据需求调整重试策略
        }
    }
    df = pd.read_parquet(GS_PATH, storage_options=storage_options)
    # 后续处理逻辑

方案3：结合多线程与多进程（CPU密集型场景） #

如果后续处理是CPU密集型任务，可采用“多线程下载+多进程处理”的架构：用线程池读取GCS文件到内存，再将数据交给进程池处理，避免多进程直接访问GCS：

from concurrent.futures import ThreadPoolExecutor, ProcessPoolExecutor
from google.cloud import storage
from io import BytesIO
import pandas as pd

def download_file():
    storage_client = storage.Client()
    bucket = storage_client.get_bucket("bucket")
    blob = bucket.blob(GS_PATH)
    byte_stream = BytesIO()
    blob.download_to_file(byte_stream)
    byte_stream.seek(0)
    return byte_stream

def process_chunk(byte_stream):
    df = pd.read_parquet(byte_stream)
    # CPU密集型处理逻辑

num_files = 1000
with ThreadPoolExecutor(max_workers=4) as download_executor, \
     ProcessPoolExecutor(max_workers=2) as process_executor:
    download_futures = [download_executor.submit(download_file) for _ in range(num_files)]
    for dl_future in download_futures:
        byte_stream = dl_future.result()
        process_executor.submit(process_chunk, byte_stream)

方案4：优化任务提交逻辑 #

原代码中无限循环提交1000个任务会导致资源过载，改为分批次提交并等待批次完成，避免瞬间产生大量请求：

num_files = 1000
max_workers = 2
with ProcessPoolExecutor(max_workers=max_workers) as executor:
    while num_files > 0:
        batch = min(max_workers, num_files)
        futures = [executor.submit(read_chunk, *args) for _ in range(batch)]
        # 等待当前批次任务完成
        for future in futures:
            future.result()
        num_files -= batch

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