Python pandas读取大型CSV文件时如何添加进度条？

核心思路是利用pandas原生分块读取能力配合tqdm实现进度跟踪，不改变原有读取逻辑，性能损耗极低。

依赖导入 #

import pandas as pd
import numpy as np
from tqdm import tqdm

辅助工具函数 #

用于提前统计文件总行数，计算进度条总长度：

def get_file_total_lines(filepath: str, encoding: str = 'iso-8859-1') -> int:
    """统计CSV文件总行数，用于计算读取进度"""
    with open(filepath, 'r', encoding=encoding) as f:
        return sum(1 for _ in f)

修改后的完整读取函数 #

仅替换原有全量读取逻辑，其余业务规则完全保留：

def read_from_csv(filepath: str,
                  sep: str = ",",
                  header_line: int = 43,
                  skip_rows: int = 48,
                  chunk_size: int = 100000) -> pd.DataFrame:
    """Reads a csv file at filepath containing the vehicle trip data and 
    performs a number of formatting operations
    """
    
    # 第一次调用read_csv用于获取列名，让第二次读取时就能完成类型指定，比后续强制转换类型效率更高
    df_names: pd.Index[str] = pd.read_csv(
                            filepath,
                            sep = sep,
                            header = header_line,
                            skip_blank_lines = False,
                            skipinitialspace = True,
                            index_col = False,
                            engine = 'c',
                            nrows = 0,
                            encoding = 'iso-8859-1'
    ).columns

    # "Time"和"Time_abs"列的实际名称前会有不一致的"存储组代码"前缀，因此先存储完整列名供后续重命名使用；同时我们需要将"Time_abs"识别为字符串类型，其余列识别为浮点型，该配置存入字典供第二次read_csv调用使用
    time_col =      ""
    time_abs_col =  ""
    names_dict = {}
    for name in df_names:
        if ": Time_abs" in name:
            names_dict[name] = 'str'
            time_abs_col = name
        elif ": Time" in name:
            time_col = name
        else:
            names_dict[name] = 'float'

    
    # 需要被pandas识别为空值的取值列表，其中只有"NOVALUE"是文件中实际会出现的取值，其余为默认缺省值
    na_vals = ['', '#N/A N/A', '#NA', '-1.#IND', '-1.#QNAN', '-NaN', '-nan',
               '1.#IND', '1.#QNAN', '<NA>', 'N/A', 'NA', 'NULL', 'NaN', 'n/a',
               'nan', 'null', 'NOVALUE']

    # 分块读取文件并展示进度条
    total_valid_lines = get_file_total_lines(filepath) - skip_rows
    chunks = []
    with tqdm(total=total_valid_lines, unit='行', desc='CSV读取进度') as pbar:
        for chunk in pd.read_csv(filepath,
                                   sep = sep,
                                   skiprows = skip_rows,
                                   header = 0,
                                   names = df_names,
                                   skip_blank_lines = False,
                                   skipinitialspace = True,
                                   index_col = False,
                                   engine = 'c',
                                   na_values = na_vals,
                                   dtype = names_dict,
                                   encoding = 'iso-8859-1',
                                   chunksize=chunk_size
                                   ):
            chunks.append(chunk)
            pbar.update(len(chunk))
    df = pd.concat(chunks, ignore_index=True)

    
    # 重命名"Time"和"Time_abs"列，移除前缀的存储组标识
    df.rename(columns = {time_col: "Time", time_abs_col: "Time_abs"}, 
              inplace = True)

    # 第二次调整该列类型，从字符串转为datetime类型；极少数情况下CSV中的Time_abs列只有时间没有日期，会触发报错，此时直接保留字符串类型即可
    try:
        df[defs.time_abs] = pd.to_datetime(df[defs.time_abs])
    except:
        pass

    
    # 每行末尾的多余分隔符会被识别为一个空列，在此处删除，避免调试时产生混淆
    df.drop(df.columns[-1], axis=1, inplace=True)

    # 为后续处理添加额外列
    df[defs.lowest_gear] = np.nan
    df[defs.lowest_speed] = np.nan
    for i in list(defs.second_trailer_axles_dict.values()):
        df[i] = np.nan

    return df

注意事项 #

• 可根据本机内存大小调整chunk_size参数，数值越大读取速度越快、内存占用越高，默认10万行适配大多数场景
• 统计文件行数的编码必须和读取CSV的编码保持一致，避免编码报错
• 进度条默认按行统计，可直观展示已读行数和预估剩余时间

内容的提问来源于stack exchange，提问作者Leopold Wahlbeck