如何用Pandas按可变块大小读取含连续ID的大CSV文件？

你的思路完全可行，因为数据是有序且同ID连续的，我们可以通过两种方式实现完整读取同一ID的所有条目：

方法一：预扫描文件，计算每个ID的完整行范围 #

先快速扫描CSV文件，记录每个ID对应的起始和结束行号，再根据这些范围精准读取完整的ID块，适合超大文件场景（避免加载全量数据）。

代码实现 #

import pandas as pd

def get_id_row_ranges(csv_path):
    id_ranges = {}
    current_id = None
    start_row = 1  # 表头占第0行，数据从第1行开始计数
    with open(csv_path, 'r', encoding='utf-8') as f:
        next(f)  # 跳过表头行
        for line_num, line in enumerate(f, start=1):
            row_id = line.split(',')[0].strip()
            if row_id != current_id:
                if current_id is not None:
                    id_ranges[current_id] = (start_row, line_num - 1)
                current_id = row_id
                start_row = line_num
        # 处理最后一个ID的范围
        if current_id is not None:
            id_ranges[current_id] = (start_row, line_num)
    return id_ranges

# 读取并处理每个完整ID块
csv_path = "data.csv"
id_row_map = get_id_row_ranges(csv_path)

for id_val, (start, end) in id_row_map.items():
    # 计算当前ID的条目数量
    row_count = end - start + 1
    # 读取完整的ID数据块
    df = pd.read_csv(csv_path, skiprows=start, nrows=row_count, header=0)
    # 这里替换为你的业务处理逻辑
    print(f"=== 处理ID: {id_val} ===")
    print(df)

方法二：按块读取后合并跨块的同一ID条目 #

无需预扫描文件，按初始chunksize读取后，检查相邻块是否属于同一ID，若属于则合并后处理，适合中等大小文件。

代码实现 #

import pandas as pd

csv_path = "data.csv"
initial_chunksize = 2
prev_chunk = None

for curr_chunk in pd.read_csv(csv_path, chunksize=initial_chunksize):
    if prev_chunk is not None:
        prev_last_id = prev_chunk['id'].iloc[-1]
        curr_first_id = curr_chunk['id'].iloc[0]
        
        if prev_last_id == curr_first_id:
            # 合并两个块，分离出完整的同一ID条目
            merged = pd.concat([prev_chunk, curr_chunk])
            same_id_data = merged[merged['id'] == prev_last_id]
            remaining_data = merged[merged['id'] != prev_last_id]
            
            # 处理完整的ID块
            print(f"=== 处理ID: {prev_last_id} ===")
            print(same_id_data)
            
            # 剩余非当前ID的数据作为下一轮的prev_chunk
            prev_chunk = remaining_data if not remaining_data.empty else None
        else:
            # 前一块是完整ID块，直接处理
            print(f"=== 处理ID: {prev_last_id} ===")
            print(prev_chunk)
            prev_chunk = curr_chunk
    else:
        prev_chunk = curr_chunk

# 处理最后剩余的完整ID块
if prev_chunk is not None:
    print(f"=== 处理ID: {prev_chunk['id'].iloc[0]} ===")
    print(prev_chunk)

两种方法的适用场景 #

• 预扫描法：优先用于超大CSV文件，精准控制内存占用，但需要额外一次文件扫描
• 合并块法：适合中等大小文件，无需预扫描，逻辑更灵活但需要临时处理小块数据的合并

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