dask.dataframe不能直接用pandas.read_csv逻辑切分文件，因其采用延迟计算图而非真实数据容器，仅解析头信息、估算分区并生成任务，不实际读取文件；手动按字节切分易导致行断裂引发ParserError，需依赖blocksize与sample自动对齐换行符确保分区起始为完整行。
为什么 dask.dataframe 不能直接用 pandas.read_csv 那套逻辑切分文件
因为 dask.DataFrame 本质是延迟计算的图结构，不是真实数据容器。你调用 dd.read_csv("large.csv") 时，Dask 只解析文件头、估算分区数量、生成读取任务——它根本没打开文件，更不会把整行内容加载进内存。
常见错误是试图对 dask.DataFrame 调用 .iloc[:1000] 或 .head() 后再做处理，结果发现后续 .compute() 还是全量跑；或者误以为 .repartition(npartitions=4) 能“按行均分”，其实它只按字节范围粗略切块，若某块末尾断在 CSV 行中间，就会报 ParserError: Expected 5 fields in line X, saw 7。
- 必须确保每个分区起始位置是完整行开头（即上一行换行符之后）
-
blocksize参数控制字节级切分粒度，但需配合sample和include_path等参数校准列推断 - 如果源文件无表头，
header=None必须显式传入，否则 Dask 会默认跳过第一行，导致列数错位
怎么安全地把一个 20GB CSV 拆成 8 个逻辑分区并延迟执行清洗
核心是让 Dask 知道“哪里是合法的切分点”。用 dd.read_csv 时，优先靠 blocksize + sample 自动探测，而不是手动算偏移：
import dask.dataframe as dd
<p>df = dd.read_csv(
"huge_data.csv",
blocksize="2GB", # 每个分区目标大小，Dask 会找最近的 \n 对齐
sample=250000, # 采样前 25 万字节推断 dtype 和列名
dtype={"user_id": "string"}, # 显式指定易出错列类型，避免后期 infer 失败
assume_missing=True # 遇到空字段不报错，转为 NaN
)此时 df 是纯延迟对象，len(df) 都不会触发计算。只有调用 .compute()、.to_parquet() 或 .persist() 才真正调度任务。
立即学习“Python免费学习笔记（深入）”；
- 检查分区数：
df.npartitions—— 如果远小于预期（比如设了blocksize="1MB"却只有 1 分区），说明文件太小或换行符太少，改用split_every或手动dd.from_delayed - 不要对
df直接用df["col"].str.contains(...)后立刻.compute()，这会让整个图降级为单线程 Pandas；应先链式写完所有变换，最后统一.compute() - 若需按业务键（如日期）重分区，用
.set_index("date", sorted=True)比.repartition更稳，前提是 date 列已排序且无缺失
延迟执行时遇到 KilledWorker 或内存爆掉怎么办
不是 Dask 分布式调度器的问题——单机模式下 KilledWorker 通常意味着某个分区处理时内存超限，被系统 OOM killer 杀掉。根本原因常是：某列含超长文本未截断、字符串操作未向量化、或 .apply 里用了非 Dask-aware 函数。
- 禁用
.apply(func, axis=1)，改用.map_partitions(lambda part: part.assign(new_col=part.x.str.slice(0, 50))) - 字符串操作务必用
.str访问器（它是 Dask 实现的），别用part["text"].apply(lambda x: x.upper()) - 用
client.run(lambda: len(gc.get_objects()))（需启动Client）查各 worker 堆内存对象数，快速定位泄漏点 - 临时缓解：启动时加
memory_limit="4GB"并设processes=False强制线程模式，避开进程间内存复制开销
切割后想局部验证逻辑，又不想跑全量
最可靠的方式是提取一个真实分区做子图执行：df.get_partition(0).compute()。注意这不是“前 N 行”，而是第 0 个字节块对应的数据子集，可能含几千行也可能上万行，取决于原始文件换行密度。
如果只想测某段清洗逻辑（比如时间格式转换），推荐用 dd.from_pandas 构造小样本，再套同名方法验证：
import pandas as pd
import dask.dataframe as dd
<h1>构造带典型脏数据的测试 Pandas DataFrame</h1><p>test_pd = pd.DataFrame({"ts": ["2023/01/01 10:30", "invalid", "2023-01-02T14:22:01"]})
test_dd = dd.from_pandas(test_pd, npartitions=1)</p><h1>复用你的清洗函数</h1><p>def clean_ts(x):
return pd.to_datetime(x, errors="coerce")</p><p>result = test_dd["ts"].map_partitions(clean_ts).compute()</p>这样既保证函数签名和行为一致，又绕开了大文件 IO 开销。等验证通过，再把同一函数丢进真实 df["ts"].map_partitions(clean_ts) 流水线里。
分区对齐和类型稳定性比“切得快”重要得多；很多失败不是因为数据太大，而是某次 astype("int") 遇到空值崩在第 7 个分区，而前 6 个都顺利跑完了。
