如何高效解析大型CSV并为各列正确转换数据类型？

背景说明 #

我需要解析多个超大型CSV文件，总数据量达6,359,008行，磁盘占用367.105 MiB，若不复用已加载信息，仅加载原始数据就会占用2.9GiB内存，性能优化迫在眉睫。这些CSV的列数和数据类型各不相同，但单文件内列数统一、同列字段类型一致，支持int/float/bool/str四种类型；文件中存在大量缺失字段（表现为,,），加载后需转为None；部分列全是相同的无效数据，需要跳过。

当前我通过传入数据类型列表实现字段转换，但为了兼容无类型转换的场景，不得不写一个冗余的string dummy函数，同时还面临内存占用过高的问题——这些数据是GeoLite2 IP地理数据库的文本转储，计划合并到SQLite3表中，且重复率极高，需要用唯一组合的索引替代重复数据存储。

当前实现代码 #

import csv
from typing import Generator

def string(s: str) -> str:
    return s

def parse_csv(path: str, types: list) -> Generator:
    with open(path, mode='r', encoding='utf8') as file:
        reader = csv.reader(file)
        if len(next(reader)) != len(types):
            raise ValueError('number of columns is different from supplied number of types')
        
        functions = [(i, func) for i, func in enumerate(types) if func]
        for row in reader:
            yield [func(e) if (e := row[i]) else None for i, func in functions]

对应的类型配置：

# 大型文件类型配置
IP_CITY_TYPES = [string, int, int, int, bool, bool, string, float, float, int]

IP_COUNTRY_TYPES = [
    string,
    int,
    int,
    int,
    bool,
    bool,
]

IP_ASN_TYPES = [string, int, string]

# 小型文件类型配置
CITIES_TYPES = [
    int,
    None,
    string,
    string,
    string,
    string,
    string,
    string,
    string,
    string,
    string,
    int,
    string,
    bool,
]

COUNTRIES_TYPES = [
    int,
    None,
    string,
    string,
    string,
    string,
    bool,
]

重复数据处理逻辑：

TABLE = []
GROUPS = {}
for row in parse_csv('/path/to/file', DATATYPES):
    network, *group = row
    start, end = parse_network(network)
    TABLE.append(((start, end), GROUPS.setdefault(tuple(group), len(GROUPS))))

更优的类型转换策略 #

1. 移除冗余dummy函数，直接用内置类型 #

不需要单独定义string函数，直接用Python内置的str类型替代，同时优化bool类型的转换逻辑（CSV中的布尔值通常是字符串格式，比如'1'/'0'/'True'/'False'，需要做特殊处理）：

修改后的解析函数：

import csv
from typing import Generator

def parse_csv(path: str, types: list) -> Generator:
    with open(path, mode='r', encoding='utf8') as file:
        reader = csv.reader(file)
        header_len = len(next(reader))
        if header_len != len(types):
            raise ValueError('列数与提供的类型数量不匹配')
        
        # 过滤需要跳过的列（类型为None的项）
        type_mapping = [(i, func) for i, func in enumerate(types) if func is not None]
        
        for row in reader:
            processed = []
            for idx, func in type_mapping:
                val = row[idx]
                if not val:
                    processed.append(None)
                else:
                    # 处理布尔值的字符串格式
                    if func is bool:
                        processed.append(val.strip().lower() in ('1', 'true'))
                    else:
                        processed.append(func(val))
            yield processed

对应的类型配置直接替换为内置类型：

IP_CITY_TYPES = [str, int, int, int, bool, bool, str, float, float, int]

IP_COUNTRY_TYPES = [
    str,
    int,
    int,
    int,
    bool,
    bool,
]

IP_ASN_TYPES = [str, int, str]

CITIES_TYPES = [
    int,
    None,
    str,
    str,
    str,
    str,
    str,
    str,
    str,
    str,
    str,
    int,
    str,
    bool,
]

COUNTRIES_TYPES = [
    int,
    None,
    str,
    str,
    str,
    str,
    bool,
]

2. 用Pandas分块处理，兼顾性能与内存 #

如果可以引入第三方库，Pandas的read_csv内置了大量优化，适合超大型文件处理：

• 用dtype指定列类型，na_values=['']自动将空字符串转为NaN（后续可转为None）
• usecols过滤无效列，chunksize分块加载，避免一次性占用大量内存

示例代码：

import pandas as pd

def get_usecols_and_dtype(types_list):
    # 提取需要保留的列索引和对应类型
    usecols = [i for i, t in enumerate(types_list) if t is not None]
    dtype = {i: t for i, t in enumerate(types_list) if t is not None}
    return usecols, dtype

# 以IP_CITY文件为例
usecols, dtype = get_usecols_and_dtype(IP_CITY_TYPES)

# 分块读取，每块10000行
for chunk in pd.read_csv(
    'ip_city.csv',
    usecols=usecols,
    dtype=dtype,
    na_values=[''],
    chunksize=10000
):
    # 将NaN转为None
    chunk = chunk.where(pd.notnull(chunk), None)
    # 逐行处理数据
    for _, row in chunk.iterrows():
        network, *group = row.tolist()
        start, end = parse_network(network)
        # 重复数据处理逻辑
        group_key = tuple(group)
        TABLE.append(((start, end), GROUPS.setdefault(group_key, len(GROUPS))))

3. 自动检测无效列，减少手动配置 #

可以提前读取少量样本行，自动识别全相同值的无效列，无需手动在类型列表中标记None：

def detect_useless_columns(path, sample_size=100):
    with open(path, 'r', encoding='utf8') as f:
        reader = csv.reader(f)
        next(reader)  # 跳过表头
        sample_rows = []
        try:
            for _ in range(sample_size):
                sample_rows.append(next(reader))
        except StopIteration:
            pass
        
        useless_cols = []
        col_count = len(sample_rows[0]) if sample_rows else 0
        for col_idx in range(col_count):
            values = {row[col_idx] for row in sample_rows}
            if len(values) == 1:
                useless_cols.append(col_idx)
        return useless_cols

# 使用示例：检测无效列后，在解析时跳过
useless_cols = detect_useless_columns('cities.csv')
# 修改parse_csv函数，在type_mapping中过滤这些列

4. 优化重复数据存储，进一步压缩内存 #

当前用字典存储唯一组合的方式已经高效，还可以用collections.defaultdict简化逻辑，或者用Pandas的Categorical类型自动生成索引，减少内存占用：

from collections import defaultdict

GROUPS = defaultdict(int)
next_index = 0
TABLE = []

for row in parse_csv('/path/to/file', DATATYPES):
    network, *group = row
    start, end = parse_network(network)
    group_key = tuple(group)
    if group_key not in GROUPS:
        GROUPS[group_key] = next_index
        next_index += 1
    TABLE.append(((start, end), GROUPS[group_key]))

总结 #

优先用内置类型替代冗余的dummy函数，优化布尔值的转换逻辑；如果允许引入第三方库，Pandas分块处理是超大型CSV的最优选择，既能提升类型转换性能，又能严格控制内存占用；自动检测无效列可减少手动配置的工作量；针对高重复率数据，用字典或类别类型存储唯一组合索引，能大幅降低内存消耗。

内容的提问来源于stack exchange，提问作者Ξένη Γήινος