使用Dask(fsspec)读取合并多压缩包中同名CSV文件遇问题

问题原因 #

你遇到的问题源于Dask对zip://格式URL的通配符解析逻辑和预期不符：

• zip://a.csv::foo*.zip：Dask不会将foo*.zip的通配符展开为所有匹配的压缩包，只会匹配第一个符合条件的zip文件，因此仅读取了第一个包里的a.csv。
• foo*.zip::a.csv：这种写法会将每个zip视为虚拟目录，Dask会遍历每个zip内的所有文件，而非仅筛选a.csv，所以会读取所有CSV文件。

解决方案 #

方案1：手动生成路径列表后合并 #

先获取所有匹配的zip文件路径，逐个构建指定CSV的读取URL，再用dd.concat合并结果：

import dask.dataframe as dd
import glob

# 获取所有foo开头的zip文件路径
zip_paths = glob.glob("foo*.zip")
# 为每个zip生成对应a.csv的读取路径
csv_paths = [f"zip://a.csv::{path}" for path in zip_paths]
# 读取所有文件并合并
dfa = dd.concat([dd.read_csv(p, delimiter=";", header=0, index_col=False) for p in csv_paths])

方案2：使用Dask Bag处理 #

如果需要更灵活的文件处理逻辑，可以用Dask Bag遍历zip文件，读取指定CSV后转为DataFrame：

import dask.bag as db
import dask.dataframe as dd
import glob
import pandas as pd
from zipfile import ZipFile

def extract_a_csv(zip_path):
    with ZipFile(zip_path) as zf:
        with zf.open("a.csv") as f:
            return pd.read_csv(f, delimiter=";", header=0, index_col=False)

# 生成zip文件序列并处理
zip_bag = db.from_sequence(glob.glob("foo*.zip"))
dfa = zip_bag.map(extract_a_csv).to_dataframe()

两种方法都能正确读取所有zip包中的目标CSV并合并，且符合Dask内存友好的特性，适合处理无法存入内存的大文件。

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