Python实战用Pandas快速切分千万级NGSIM交通数据集附完整代码处理千万行级别的交通轨迹数据是很多数据工程师和交通研究者在入门时遇到的第一个“拦路虎”。当你兴致勃勃地下载了著名的NGSIM数据集准备大展拳脚时一个简单的pd.read_csv()操作就可能让你的16GB内存电脑瞬间卡死进度条纹丝不动。这不仅仅是内存不足的问题更关乎后续分析、模型训练的效率和可行性。将庞大的数据集进行合理切分是处理这类数据的第一步也是将数据从“不可用”变为“可用”的关键。NGSIM下一代仿真数据集作为微观交通行为研究的基石包含了超过1100万条高精度车辆轨迹记录。对于交通流分析、驾驶行为建模乃至自动驾驶算法开发它都是不可或缺的燃料。然而其原始数据往往以单个巨型CSV或文本文件的形式提供直接加载到内存中进行操作对绝大多数个人计算机和工作站来说都是不现实的。本文将从一个实战者的角度出发不仅教你如何用Pandas高效、安全地将这个千万级数据集“化整为零”更会深入探讨在不同场景下的切分策略、内存优化技巧以及切分后如何优雅地进行合并与后续处理。我们会避开那些教科书式的简单循环分享真正在工程实践中验证过的代码和思路。1. 理解NGSIM数据集为什么必须切分在动手写代码之前我们有必要先搞清楚面对的是什么以及为什么常规方法会失效。NGSIM数据集并非一个单一的文件它实际上包含了在US-101和I-80等高速路段采集的多个15分钟片段的数据。即便是一个片段其数据量也极为庞大。数据规模与结构挑战一个典型的NGSIM数据文件例如us-101.csv可能包含数十个字段从车辆ID、时间戳、全局坐标X/Y到速度、加速度、与前车的距离等。以1180万条记录为例假设每条记录有20个字段多为浮点数或整数粗略估算其内存占用可能达到11,800,000 行 * 20 列 * 8 字节/浮点数 ≈ 1.888 GB这还仅仅是数据本身加载为NumPy数组的理论下限。Pandas DataFrame由于存在索引、数据类型转换object类型会占用更多空间、以及内部优化结构实际内存占用可能是这个数值的2到3倍轻松突破4GB。对于大多数开发环境来说这已经接近或超过了可用物理内存的极限导致操作系统开始使用速度缓慢的虚拟内存硬盘交换空间从而引发程序卡顿甚至崩溃。切分的核心目的因此对NGSIM数据集进行切分首要目的就是解决内存瓶颈使其能够在有限资源下被处理。其次切分能带来诸多好处并行处理将数据分成多个块后可以分发到多台机器或多个进程进行并行计算显著提升处理速度。增量学习对于机器学习任务可以按块加载数据进行增量训练无需一次性载入全部数据。版本管理与备份小文件更易于管理、备份和共享。针对性分析可以只加载特定时间段、特定路段或特定车辆的数据进行分析提高效率。注意切分不是简单的“一刀切”。我们需要根据后续分析的需求来选择切分维度是按时间、按车辆ID、还是按固定行数不同的选择会直接影响后续操作的复杂度。2. 环境准备与高效读取策略工欲善其事必先利其器。在开始切分之前确保你的Python环境已经就绪并掌握一些高效读取大型CSV文件的技巧。基础环境配置首先确保安装了必要的数据处理库。建议使用conda或pip创建一个干净的虚拟环境。# 使用conda创建环境可选 conda create -n ngsi m_processing python3.9 conda activate ngsi m_processing # 安装核心库 pip install pandas numpy # 可选安装用于加速IO的库 pip install pyarrow高效读取的第一次接触只窥一斑在切分前我们至少需要知道数据的结构列名。直接读取整个文件是不可取的。Pandas的read_csv函数提供了nrows参数允许我们只读取前几行来获取元信息。import pandas as pd from pathlib import Path # 定义数据路径 data_path Path(./data/NGSIM_US-101.csv) # 仅读取前10行来获取列名和数据类型预览 df_sample pd.read_csv(data_path, nrows10) print(数据列名, df_sample.columns.tolist()) print(数据预览) print(df_sample.head()) print(\n数据类型) print(df_sample.dtypes)执行这段代码你会立刻看到数据的字段例如Vehicle_ID,Frame_ID,Global_Time,Local_X,Local_Y等。了解数据类型至关重要因为正确的数据类型能节省大量内存。例如如果Vehicle_ID是整数但被读成了浮点数内存占用就会翻倍。内存优化指定数据类型Dtype这是提升大文件处理效率的关键一步。我们可以通过dtype参数为每一列指定最节省内存的数据类型。# 根据预览为各列指定优化的数据类型 dtype_spec { Vehicle_ID: int32, # 车辆ID通常用32位整数足够 Frame_ID: int32, Global_Time: int64, # 时间戳可能很大用64位 Local_X: float32, # 坐标用32位浮点精度足够 Local_Y: float32, v_Length: float32, v_Width: float32, v_Class: int8, # 车型分类类别少用8位整数 v_Vel: float32, v_Acc: float32, Lane_ID: int8, # ... 根据实际字段补充 } # 再次读取样本验证数据类型 df_sample_typed pd.read_csv(data_path, nrows10, dtypedtype_spec) print(优化后内存占用样本:, df_sample_typed.memory_usage(deepTrue).sum() / 1024**2, MB)通过指定dtype我们可能将内存占用降低50%甚至更多。对于千万行数据这意味着一到两个GB的节省往往是成功加载与否的关键。3. 核心切分策略与代码实现现在进入核心环节如何将庞大的文件切成小块。我们将探讨两种最实用的策略按固定行数切分和按逻辑分组如车辆ID切分。3.1 策略一按固定行数切分通用方法这是最直接的方法尤其适用于后续需要随机访问或均匀分布处理的任务。思路是分批读取数据每一批达到指定行数后就保存为一个新文件。基础版本使用skiprows和nrows下面的代码实现了以100万行为一个文件进行切分。它巧妙地利用skiprows参数跳过已读取的行用nrows控制每次读取的量。import pandas as pd from pathlib import Path def split_csv_by_rows(file_path, chunk_size1_000_000, output_dirNone): 按固定行数切分大型CSV文件。 参数 file_path: 原始CSV文件路径 chunk_size: 每个切分文件包含的行数 output_dir: 输出目录默认为原文件目录下的split_files文件夹 file_path Path(file_path) if output_dir is None: output_dir file_path.parent / fsplit_{file_path.stem} output_dir Path(output_dir) output_dir.mkdir(parentsTrue, exist_okTrue) print(f开始切分文件: {file_path}) print(f目标目录: {output_dir}) print(f切分大小: {chunk_size:,} 行/文件) # 第一步获取列名 # 读取文件头假设第一行是列名 with open(file_path, r) as f: header f.readline().strip() # 第二步计算总行数不包括标题行 # 这是一个耗时的操作对于超大文件可以跳过或估算 print(正在计算文件总行数...) total_rows 0 with open(file_path, r) as f: # 跳过标题行 next(f) for line in f: total_rows 1 print(f文件总数据行数: {total_rows:,}) # 第三步循环读取和写入 chunk_no 0 rows_processed 0 while rows_processed total_rows: # 计算本次读取的起始行和行数 skip_rows 1 rows_processed # 加1以跳过标题行 nrows min(chunk_size, total_rows - rows_processed) # 使用低内存模式读取块 # 注意此处再次指定dtype可以进一步优化但为简化我们使用自动推断 # 在实际生产中强烈建议传入预先定义的dtype_spec chunk pd.read_csv(file_path, skiprowsskip_rows, nrowsnrows, headerNone, # 因为我们跳过了标题行 namesheader.split(,), # 手动指定列名 low_memoryFalse) # 关闭低内存模式以加速 # 生成输出文件名 start_row rows_processed 1 end_row rows_processed len(chunk) output_file output_dir / f{file_path.stem}_part_{chunk_no:04d}_rows_{start_row}_{end_row}.csv # 保存块 chunk.to_csv(output_file, indexFalse) print(f已保存: {output_file.name} (行 {start_row:,} - {end_row:,})) # 更新计数器 rows_processed len(chunk) chunk_no 1 print(f\n切分完成共生成 {chunk_no} 个文件。) # 使用示例 if __name__ __main__: # 替换为你的实际文件路径 data_file ./data/NGSIM_US-101.csv split_csv_by_rows(data_file, chunk_size1_000_000)代码解析与优化点总行数计算代码先计算总行数以确定循环次数。对于超大型文件这一步可能较慢。如果文件行数已知或可以估算可以跳过此步改用try-except捕获读取结束。内存管理通过skiprows和nrows我们确保每次只有一小块数据100万行被加载到内存中处理完后立即释放。文件命名文件名包含了起始和结束行号便于后续追溯和按需加载。low_memory参数对于字段类型混合的文件Pandas的low_memoryTrue默认会分块推断类型可能更安全但稍慢。如果我们已通过样本明确了数据类型使用low_memoryFalse并指定dtype会更快。3.2 策略二按车辆ID分组切分业务驱动方法在交通轨迹分析中更常见的需求是按车辆进行分析。例如研究每辆车的跟驰行为、换道决策。这时按车辆ID切分比按行数切分更有意义因为每个文件包含了一辆或几辆车的完整轨迹数据在业务上是完整的。实现思路这种方法无法简单通过skiprows实现因为同一车辆的数据行在原始文件中是交错分布的。我们需要按块读取数据。在内存中对每个块按Vehicle_ID进行分组。将不同块中属于同一车辆的数据追加到同一个输出文件中。这涉及到更复杂的状态管理需要知道哪些车辆的数据已经写入到哪个文件。一个更实用的简化方法是先按行数切分成中间文件再对每个中间文件进行按车辆ID的聚合重组。但这样会产生大量IO。更高效的按车辆切分使用迭代器与字典缓存下面是一种更精巧的方法它使用Pandas的chunksize参数进行流式读取并维护一个字典来按车辆ID写入数据import pandas as pd from pathlib import Path import csv def split_csv_by_vehicle(file_path, output_dir, vehicles_per_file100): 按车辆ID切分CSV文件每个文件包含指定数量的车辆数据。 注意此方法适用于车辆ID分布相对均匀的情况且需要足够内存缓存部分数据。 file_path Path(file_path) output_dir Path(output_dir) output_dir.mkdir(parentsTrue, exist_okTrue) # 用于跟踪车辆ID到文件句柄的映射 vehicle_to_writer {} # 用于跟踪每个输出文件包含了哪些车辆 file_to_vehicles {} current_file_index 0 vehicles_in_current_file set() # 获取列名 with open(file_path, r) as f: header f.readline().strip() fieldnames header.split(,) # 创建第一个输出文件 current_output_file output_dir / f{file_path.stem}_vehicles_001.csv f_current open(current_output_file, w, newline) writer_current csv.DictWriter(f_current, fieldnamesfieldnames) writer_current.writeheader() file_to_vehicles[current_output_file] vehicles_in_current_file print(f开始按车辆切分每文件约 {vehicles_per_file} 辆车...) # 使用chunksize流式读取 chunk_size 50000 # 每次读取5万行 for chunk in pd.read_csv(file_path, chunksizechunk_size, dtype{Vehicle_ID: int32}): # 按Vehicle_ID分组 grouped chunk.groupby(Vehicle_ID) for vehicle_id, group_df in grouped: vehicle_id int(vehicle_id) # 确保为标量 # 检查该车辆是否已有对应的写入器 if vehicle_id not in vehicle_to_writer: # 如果当前文件车辆数已满创建新文件 if len(vehicles_in_current_file) vehicles_per_file: f_current.close() current_file_index 1 current_output_file output_dir / f{file_path.stem}_vehicles_{current_file_index1:03d}.csv f_current open(current_output_file, w, newline) writer_current csv.DictWriter(f_current, fieldnamesfieldnames) writer_current.writeheader() vehicles_in_current_file set() file_to_vehicles[current_output_file] vehicles_in_current_file # 将此车辆关联到当前写入器 vehicle_to_writer[vehicle_id] writer_current vehicles_in_current_file.add(vehicle_id) # 获取该车辆对应的写入器 writer vehicle_to_writer[vehicle_id] # 将DataFrame转换为字典列表并写入 for _, row in group_df.iterrows(): writer.writerow(row.to_dict()) # 关闭所有文件句柄 f_current.close() print(f切分完成共生成 {current_file_index 1} 个车辆分组文件。) # 打印每个文件包含的车辆ID范围示例 for f_path, v_set in list(file_to_vehicles.items())[:5]: # 只打印前5个 if v_set: print(f {f_path.name}: 车辆ID {min(v_set)} 到 {max(v_set)}) # 使用示例注意此方法对内存和IO要求较高适用于车辆数不是特别多的情况 # split_csv_by_vehicle(./data/NGSIM_US-101.csv, ./output/vehicle_split, vehicles_per_file200)提示按车辆切分的方法内存消耗和IO操作更密集因为需要同时在内存中维护多个文件句柄和映射关系。对于车辆数极多如数万辆的情况可能需要调整vehicles_per_file参数或者采用先按行切分、再对每个小文件按车辆合并的两阶段策略。4. 高级技巧内存映射与Dask框架当Pandas单机处理显得力不从心时我们可以借助更强大的工具。使用Pandas的memory_map选项对于存储在SSD上的文件启用内存映射可以提升读取速度尤其是在多次随机访问时。# 在read_csv中启用内存映射适用于Linux/Unix系统Windows可能效果不佳 df_chunk pd.read_csv(file_path, nrowschunk_size, memory_mapTrue)memory_mapTrue会告诉Pandas使用操作系统提供的内存映射文件功能这可以减少内核空间和用户空间之间的数据复制对于非常大的文件可能有一定提升。但请注意它并不减少总的内存占用。拥抱Dask并行化与核外计算Dask是一个用于并行计算的灵活库它可以处理大于内存的数据集。Dask的DataFrame接口与Pandas非常相似但它在幕后将大型数据集分割成许多小的Pandas DataFrame并跨多个CPU核心或甚至集群进行计算。import dask.dataframe as dd # 使用Dask直接读取大型CSV惰性加载不立即读入内存 dask_df dd.read_csv(./data/NGSIM_US-101.csv, dtypedtype_spec) # 使用之前定义的dtype # 查看数据概览触发部分计算 print(dask_df.head()) print(f数据集分区数: {dask_df.npartitions}) print(f预估行数: {len(dask_df):,}) # 使用Dask进行切分并保存为多个CSV # 这将根据数据大小和块大小自动创建多个分区并将每个分区保存为独立的CSV文件 output_path_pattern ./output/dask_split/part_*.csv dask_df.to_csv(output_path_pattern, indexFalse, single_fileFalse) print(Dask并行切分任务已提交。)使用Dask的好处是你可以用类似Pandas的语法操作远超内存的数据并且计算会自动并行化。to_csv操作会为每个分区生成一个文件如part_0.csv,part_1.csv。Dask会自动管理分区大小你也可以通过blocksize参数来控制。两种策略对比特性Pandas分块读取 (skiprows/nrows)Dask DataFrame学习曲线较低基于熟悉的Pandas API中等需要理解惰性计算和任务图内存要求需手动控制块大小单块数据需能放入内存自动核外计算可处理远大于内存的数据并行能力需手动实现如multiprocessing内置并行自动利用多核适用场景一次性脚本简单的顺序切分复杂的数据处理管道需要多次转换和聚合IO模式顺序读取可并行读取如果文件支持对于“一次性将大文件切成小文件”这个任务如果逻辑简单Pandas分块方法足够且直观。但如果后续还有复杂的过滤、聚合、计算等步骤那么从开始就使用Dask可能会更高效。5. 切分后的数据管理、合并与验证文件切分完成后工作只完成了一半。如何有效地管理和使用这些碎片化的数据是下一个挑战。文件命名与元数据记录良好的文件命名规范至关重要。建议在文件名中包含数据集名称如NGSIM_US101切分逻辑如by_row,by_vehicle序列号或范围如part_001,rows_1_1000000,vehicles_1_200切分日期或版本号同时创建一个元数据文件如manifest.json记录切分信息非常有用{ dataset: NGSIM_US-101, split_method: by_rows, chunk_size: 1000000, total_chunks: 12, chunks: [ {file: NGSIM_US101_part_01_rows_1_1000000.csv, rows: [1, 1000000]}, {file: NGSIM_US101_part_02_rows_1000001_2000000.csv, rows: [1000001, 2000000]}, // ... ], created_at: 2023-10-27T10:30:00Z }按需合并不是所有时候都需要“合体”切分后很多分析任务并不需要完整的数据集。例如统计每辆车的平均速度可以按车辆ID分组处理每个文件再合并结果。分析特定时间段的交通流可以先筛选出对应时间范围的文件进行处理。只有在进行全数据集范围的排序、去重或需要连续时间序列分析时才需要合并。合并小文件时可以再次使用Pandas的增量读取和追加写入def merge_chunks(chunk_files, output_file): 将多个切分文件合并为一个。 output_file Path(output_file) # 读取第一个文件获取列名和写入头 first_chunk pd.read_csv(chunk_files[0], nrows0) first_chunk.to_csv(output_file, indexFalse) # 追加后续文件跳过标题行 for chunk_file in chunk_files: # 使用headerNone, names列名的方式读取避免重复标题行问题 chunk pd.read_csv(chunk_file, header0) # 假设每个文件都有标题行 # 以追加模式写入不写列名 chunk.to_csv(output_file, modea, indexFalse, headerFalse) print(f已追加: {chunk_file}) print(f合并完成输出至: {output_file})数据完整性验证切分和合并后必须验证数据没有丢失或错乱。行数校验确保所有切分文件的行数之和等于原始文件行数。关键字段校验随机抽样检查某些车辆ID的轨迹在切分前后是否连续、完整。统计量对比计算原始文件和合并后文件的某些关键字段如总车辆数、时间范围、速度均值是否一致。def validate_split(original_file, chunk_files, sample_vehicle_ids[1, 100, 500]): 验证切分数据的完整性。 # 1. 行数校验 total_rows_original 0 with open(original_file, r) as f: total_rows_original sum(1 for line in f) - 1 # 减去标题行 total_rows_chunks 0 for cf in chunk_files: with open(cf, r) as f: total_rows_chunks sum(1 for line in f) - 1 print(f原始文件行数: {total_rows_original:,}) print(f切分文件总行数: {total_rows_chunks:,}) print(f行数是否一致: {total_rows_original total_rows_chunks}) # 2. 抽样车辆轨迹连续性检查示例需根据实际数据结构调整 # 此检查较为复杂通常需要加载相关车辆的所有数据片段进行比对 # 这里仅示意流程 for vid in sample_vehicle_ids: print(f\n检查车辆 ID {vid} 的轨迹...) # 收集该车辆在所有切分文件中的数据 vehicle_data [] for cf in chunk_files: # 注意这里应使用高效查询例如将文件读入数据库或用Dask # 为简化假设我们有一个函数能快速从CSV中提取特定车辆的数据 # df_chunk pd.read_csv(cf) # vehicle_chunk df_chunk[df_chunk[Vehicle_ID] vid] # vehicle_data.append(vehicle_chunk) pass # 合并并检查时间戳是否连续递增等 # ...6. 实战案例构建一个可复用的切分管道最后我们将上述知识点整合构建一个更健壮、可配置的NGSIM数据集切分脚本。这个脚本考虑了错误处理、进度显示和配置化。 NGSIM数据集智能切分管道 支持按行数、按车辆ID切分并生成元数据报告。 import pandas as pd import numpy as np from pathlib import Path import json import argparse from datetime import datetime import sys class NGSIMSplitter: def __init__(self, data_path, output_base_dir./output): self.data_path Path(data_path) if not self.data_path.exists(): raise FileNotFoundError(f数据文件不存在: {data_path}) self.output_base Path(output_base_dir) self.output_base.mkdir(parentsTrue, exist_okTrue) self.meta_info { original_file: str(self.data_path), split_timestamp: datetime.now().isoformat(), total_rows: None, chunks: [] } def get_total_rows(self): 快速计算CSV文件行数不包括标题行。 print(正在计算文件总行数...) with open(self.data_path, rb) as f: # 这种方法比逐行读取快但假设没有带换行的字段 total_rows sum(1 for _ in f) - 1 self.meta_info[total_rows] total_rows return total_rows def split_by_rows(self, rows_per_chunk1_000_000, dtype_mapNone): 按行数切分。 print(f\n 开始按行切分每块 {rows_per_chunk:,} 行 ) total_rows self.get_total_rows() num_chunks (total_rows // rows_per_chunk) (1 if total_rows % rows_per_chunk else 0) # 创建本次切分的输出目录 run_id datetime.now().strftime(%Y%m%d_%H%M%S) output_dir self.output_base / fsplit_by_rows_{run_id} output_dir.mkdir(exist_okTrue) self.meta_info[split_method] by_rows self.meta_info[rows_per_chunk] rows_per_chunk self.meta_info[output_dir] str(output_dir) # 读取列名 with open(self.data_path, r, encodingutf-8) as f: header f.readline().strip() col_names header.split(,) chunk_no 0 rows_processed 0 # 使用tqdm显示进度条如果已安装 try: from tqdm import tqdm pbar tqdm(totaltotal_rows, desc切分进度) except ImportError: pbar None while rows_processed total_rows: skip_rows 1 rows_processed nrows min(rows_per_chunk, total_rows - rows_processed) # 读取数据块 chunk pd.read_csv(self.data_path, skiprowsskip_rows, nrowsnrows, headerNone, namescol_names, dtypedtype_map, low_memoryFalse) # 生成输出文件 start_row rows_processed 1 end_row rows_processed len(chunk) output_file output_dir / fchunk_{chunk_no:04d}_rows_{start_row}_{end_row}.parquet # 使用Parquet格式 # 保存为Parquet比CSV更小更快 chunk.to_parquet(output_file, indexFalse, compressionsnappy) # 记录元数据 self.meta_info[chunks].append({ chunk_id: chunk_no, file: output_file.name, start_row: start_row, end_row: end_row, num_rows: len(chunk) }) rows_processed len(chunk) chunk_no 1 if pbar: pbar.update(len(chunk)) else: print(f进度: {rows_processed:,} / {total_rows:,} 行) if pbar: pbar.close() # 保存元数据 meta_file output_dir / split_metadata.json with open(meta_file, w) as f: json.dump(self.meta_info, f, indent2, ensure_asciiFalse) print(f切分完成共生成 {chunk_no} 个文件于 {output_dir}) print(f元数据已保存至: {meta_file}) return output_dir def split_by_vehicle(self, vehicles_per_chunk200, sample_size10000): 按车辆ID切分简化版适用于车辆ID分布均匀的数据。 print(f\n 开始按车辆切分每块约 {vehicles_per_chunk} 辆车 ) # 注意完整实现需要考虑内存和车辆ID分布这里提供一个框架思路 # 1. 首先采样估算总车辆数和ID范围 print(正在采样数据以分析车辆ID分布...) df_sample pd.read_csv(self.data_path, nrowssample_size) unique_vehicles_sample df_sample[Vehicle_ID].unique() print(f样本中发现的唯一车辆数: {len(unique_vehicles_sample)}) # 2. 由于车辆ID在文件中是交错的更稳健的方法是 # a) 使用Dask进行groupby操作输出按车辆ID分区的Parquet数据集 # b) 或者使用数据库如DuckDB进行分组和导出 print(建议对于按车辆ID切分使用Dask或数据库工具更为高效。) print(例如使用Dask: dd.read_csv(...).to_parquet(..., partition_on[Vehicle_ID])) # 此处省略具体实现因其复杂度较高需根据具体需求定制 return None def main(): parser argparse.ArgumentParser(descriptionNGSIM大型数据集切分工具) parser.add_argument(input_file, help输入的NGSIM CSV文件路径) parser.add_argument(--method, choices[rows, vehicle], defaultrows, help切分方法: rows (按行) 或 vehicle (按车辆ID)) parser.add_argument(--chunk-size, typeint, default1000000, help按行切分时每个文件的行数 (默认: 1,000,000)) parser.add_argument(--vehicles-per-file, typeint, default200, help按车辆切分时每个文件的车辆数 (默认: 200)) parser.add_argument(--output-dir, default./split_output, help输出目录 (默认: ./split_output)) args parser.parse_args() splitter NGSIMSplitter(args.input_file, args.output_dir) # 定义优化的数据类型根据你的NGSIM数据字段调整 dtype_optimized { Vehicle_ID: int32, Frame_ID: int32, Global_Time: int64, Local_X: float32, Local_Y: float32, v_Vel: float32, v_Acc: float32, Lane_ID: int8, Preceding: int32, Following: int32, Space_Headway: float32, Time_Headway: float32 } if args.method rows: splitter.split_by_rows(rows_per_chunkargs.chunk_size, dtype_mapdtype_optimized) elif args.method vehicle: splitter.split_by_vehicle(vehicles_per_chunkargs.vehicles_per_file) else: print(未知的切分方法。) if __name__ __main__: main()这个脚本提供了命令行接口可以通过参数指定切分方式和大小。它还将输出格式从CSV换成了Parquet这是一种列式存储格式具有更好的压缩比和读取性能特别适合后续的统计分析。使用前需要安装pyarrow或fastparquet库pip install pyarrow。将数据切分并保存为Parquet格式后你甚至可以直接用Dask或Pandas配合pyarrow引擎高效地查询其中部分数据无需合并整个数据集。处理千万级数据就像操作一台精密仪器每一步都需要权衡内存、IO和计算资源。从了解数据规模开始到选择合适的数据类型、设计切分策略、利用高效工具最后进行验证和管理构成了一个完整的数据预处理工作流。本文介绍的方法并非唯一解但希望能为你提供一个坚实的起点。在实际项目中你可能会遇到更复杂的情况比如数据压缩、网络存储读取、或者需要在云环境如AWS S3中直接处理。这时Dask、Ray等分布式计算框架或者像DuckDB这样的嵌入式分析数据库会成为更强大的武器。记住核心原则始终是让数据流动起来而不是堆积在内存里。