使用Dask加载30GB大CSV文件遇阻求助

问题背景 #

尝试用Dask加载30GB CSV文件（因内存不足无法使用Pandas），设备配置为126GB RAM、32核、10GB交换空间，但始终失败；朋友用64GB RAM的Mac成功加载同一文件，推测是操作疏漏导致。以下是尝试的两种方法及报错，附修复方案：

尝试的方法与报错 #

方法1：基于内存和核心数估算chunk size #

import psutil
import os
import dask.dataframe as dd

def estimate_chunk_size() -> int:
    # 计算可用内存和核心数
    memory_available_bytes = psutil.virtual_memory().available
    num_cores = os.cpu_count()

    # 根据内存和核心数计算chunk大小
    chunk_size_bytes = int((memory_available_bytes / num_cores) / 6)

    return chunk_size_bytes

def csv_loader(csv_path: str) -> dd.DataFrame:
    try:
        # 获取文件大小
        file_size_gb = os.path.getsize(csv_path) / (1024 ** 3)
        
        # 指定数据类型
        dtype = {
            'end_date': 'object',
            ...
            'user_last_update': 'object'
        }
        
        # 估算chunk大小
        blocksize = estimate_chunk_size()

        # 用Dask加载CSV
        csv_data = dd.read_csv(csv_path, assume_missing=True, blocksize=blocksize, dtype=dtype)

        return csv_data

    except FileNotFoundError:
        # 处理文件不存在错误
        ...

    except Exception as e:
        # 处理其他异常
        ...

报错信息：

Columns (19,20,21,22,23,29,30,31,32,33,34,35,36,37,38,39,40,41,42,43,44,45,46,47,110) have mixed types.Specify dtype option on import or set low_memory=False.

^C
Process finished with exit code 137

方法2：启用Dask Client（设置最大worker数） #

import psutil
import os
import dask.dataframe as dd
from dask.distributed import Client

def estimate_chunk_size() -> int:
    # 计算可用内存和核心数
    memory_available_bytes = psutil.virtual_memory().available
    num_cores = os.cpu_count()

    # 根据内存和核心数计算chunk大小
    chunk_size_bytes = int((memory_available_bytes / num_cores) / 6)

    return chunk_size_bytes

def csv_loader(csv_path: str) -> dd.DataFrame:
    try:
        # 获取文件大小
        file_size_gb = os.path.getsize(csv_path) / (1024 ** 3)

        # 指定数据类型
        dtype = {
            'end_date': 'object',
            ...
            'user_last_update': 'object'
        }

        # 估算chunk大小
        blocksize = estimate_chunk_size()

        # 创建Dask Client并设置最大worker数
        with Client(n_workers=psutil.cpu_count(logical=True)) as client:
            # 用Dask加载CSV
            csv_data = dd.read_csv(csv_path, assume_missing=True, blocksize=blocksize, dtype=dtype)

        return csv_data

    except FileNotFoundError:
        # 处理文件不存在错误
        ...

    except Exception as e:
        # 处理其他异常
        ...

修复方案 #

1. 解决混合类型列问题 #

报错提示的列未被包含在指定的dtype字典中，导致Dask推断类型时出现冲突。两种解决方式：

• 补充所有混合类型列的dtype定义（例如将这些列设为object或对应实际类型）；
• 在dd.read_csv中添加low_memory=False参数，关闭分块类型推断的限制。

2. 调整chunk size与Worker配置 #

• 固定blocksize：放弃动态计算，直接设置blocksize='256MB'（Dask推荐范围为64MB-1GB），避免因内存计算误差导致chunk过大；
• 限制Worker数量：将n_workers设为物理核心数（16）而非逻辑核心数（32），同时指定每个Worker的内存上限，例如：

  Client(n_workers=16, memory_limit='8GB')
  

  避免过度调度引发内存溢出。

3. 修正Client生命周期问题 #

Dask Client在with块结束后会自动关闭，而返回的dd.DataFrame是惰性计算对象，后续操作会因Client关闭失败。需将Client创建放在函数外部，或确保整个数据处理流程在Client存活期间执行。

4. 处理Exit Code 137 #

该代码表示进程因内存不足被系统杀死，按上述步骤调整后，可有效控制内存占用，解决OOM问题。

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