十万行DataFrame内连接卡顿：用Dask仍死机，求合理解决方案

首先，10万行的数据集其实不算超大，但如果列数多、数据类型不合理或者Dask使用方式不对，确实会导致内存过载。结合你的代码，我整理了几个实用的优化方案：

1. 优化Dask的连接方式：避免手动拼接Unique键 #

你原来的代码通过拼接多列生成Unique字符串列再连接，这会额外占用大量内存（字符串列比原始分类列占空间多得多），而且效率低。直接用多列作为连接键更高效：

import dask.dataframe as dd

# 读取数据时直接指定必要的列和数据类型
dtypes = {
    "OrderDate": "object",  # 如果是日期格式可以改成datetime64[ns]
    "Region": "category",
    "Rep": "category",
    "Item": "category"
}

SData = dd.read_csv("KD_111.csv", dtype=dtypes)
TData = dd.read_csv("KD_111_T.csv", dtype=dtypes)

# 直接用多列做内连接，替代Unique列
merged = dd.merge(SData, TData, on=["OrderDate", "Region", "Rep", "Item"], how="inner")

# Dask是惰性执行，需要compute触发计算
result = merged.compute()

2. 调整Dask分区大小，降低单分区内存压力 #

默认的read_csv分区可能过大，导致单个分区处理时占用内存超过阈值。可以手动指定blocksize来控制分区大小（比如64MB-128MB，根据你的列数调整）：

# 按64MB分块读取，适配8GB内存的处理能力
SData = dd.read_csv("KD_111.csv", dtype=dtypes, blocksize="64MB")
TData = dd.read_csv("KD_111_T.csv", dtype=dtypes, blocksize="64MB")

3. 提前筛选必要列，减少数据量 #

如果你的CSV里有很多不需要的列，读取时只保留连接键和最终需要的列，能大幅降低内存占用：

# 只保留需要的列：连接键+业务列
keep_cols = ["OrderDate", "Region", "Rep", "Item", "Sales", "Quantity"]
SData = dd.read_csv("KD_111.csv", usecols=keep_cols, dtype=dtypes)
TData = dd.read_csv("KD_111_T.csv", usecols=keep_cols, dtype=dtypes)

4. 用Pandas分块处理替代Dask（备选方案） #

如果Dask还是出现问题，8GB内存完全可以用Pandas分块处理。先把其中一个数据集加载到内存（10万行完全没问题），再分块读取另一个数据集逐块连接：

import pandas as pd

# 先加载TData到内存，同时优化数据类型
t_df = pd.read_csv("KD_111_T.csv")
t_df[["Region", "Rep", "Item"]] = t_df[["Region", "Rep", "Item"]].astype("category")
t_df["OrderDate"] = t_df["OrderDate"].astype("object")  # 或者转datetime类型

# 分块读取SData，逐块连接
chunk_size = 10000  # 每次处理1万行，可根据内存情况调整
merged_chunks = []

for s_chunk in pd.read_csv("KD_111.csv", chunksize=chunk_size):
    # 同步优化chunk的数据类型，保证连接时类型一致
    s_chunk[["Region", "Rep", "Item"]] = s_chunk[["Region", "Rep", "Item"]].astype("category")
    s_chunk["OrderDate"] = s_chunk["OrderDate"].astype("object")
    
    # 内连接当前chunk和TData
    merged_chunk = pd.merge(s_chunk, t_df, on=["OrderDate", "Region", "Rep", "Item"], how="inner")
    merged_chunks.append(merged_chunk)

# 合并所有分块结果
final_result = pd.concat(merged_chunks, ignore_index=True)

5. 额外优化：清理内存碎片 #

如果你的代码运行中内存占用持续升高，可以手动触发垃圾回收，或者避免不必要的变量留存：

import gc

# 在关键步骤后清理内存
del SData, TData
gc.collect()

这些方案都是针对8GB内存优化的，核心思路是减少不必要的数据占用和拆分任务降低单步内存压力，你可以根据自己的实际数据情况选择最合适的方式。

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