Python中不加载整个文件到内存读取指定行的方法

绝对有办法解决这个问题！9万行2万列的CSV直接全加载确实会吃光内存，下面分两种情况给你方案——不换格式的应急办法，以及换格式的长期优化方案：

这种方法不需要改动原文件，完全靠逐行遍历筛选，内存占用极低，只保留你需要的行数据。

用Python内置csv模块（最省内存） #

这是最稳妥的方案，csv模块天生就是逐行处理的，不会把整个文件塞进内存。你只需要把目标行的索引存成一个集合（查找更快），然后遍历每一行，匹配到就保留，甚至可以找到所有目标行后提前退出，节省时间：

import csv

# 把你要读取的行索引存在集合里（注意：行号从0开始，第一行是表头的话，数据行从1开始）
target_row_indices = {100, 2000, 50000}
selected_rows = []

with open('your_large_file.csv', 'r', newline='') as csv_file:
    reader = csv.reader(csv_file)
    for row_num, row in enumerate(reader):
        if row_num in target_row_indices:
            selected_rows.append(row)
            # 所有目标行都找到了就提前结束，不用再读后面的内容
            if len(selected_rows) == len(target_row_indices):
                break

# 现在selected_rows里就是你要的行数据了

用Pandas的skiprows参数（更便捷） #

如果你习惯用Pandas处理数据，也可以用read_csv的skiprows参数，传入一个lambda函数来跳过不需要的行。本质上也是逐行遍历，但会帮你把结果整理成DataFrame：

import pandas as pd

target_row_indices = {100, 2000, 50000}
# skiprows返回True就跳过该行，所以我们只保留在目标集合里的行
df = pd.read_csv('your_large_file.csv', skiprows=lambda x: x not in target_row_indices)

注意：这里的行号也是从0开始的，如果你的CSV有表头，记得根据实际情况调整目标索引（比如表头是行0，数据行从1开始）。

如果需要频繁读取特定行/列，换成支持随机访问的格式会更高效，不用每次都遍历整个文件。推荐这几种：

Parquet（首推！） #

Parquet是列存储格式，压缩率极高（比CSV小很多），而且支持直接按行索引读取，速度超快。只需要把CSV转成Parquet一次，之后读取就很方便：

第一步：把CSV转成Parquet #

如果原文件太大，没法一次性加载到内存，可以分块读取再合并写入：

import pandas as pd

# 分块读取CSV，每块1万行
chunk_size = 10000
chunks = []
for chunk in pd.read_csv('your_large_file.csv', chunksize=chunk_size):
    chunks.append(chunk)

# 合并所有块并写入Parquet
full_df = pd.concat(chunks)
full_df.to_parquet('your_large_file.parquet')

第二步：读取特定行 #

转完之后，直接指定行索引就能读取，不用遍历整个文件：

import pandas as pd

# 直接读取指定行，秒出结果
selected_df = pd.read_parquet('your_large_file.parquet', rows=[100, 2000, 50000])

SQLite数据库（适合复杂查询） #

如果需要经常做条件查询（不是只按行号），把数据导入SQLite数据库是个好选择。SQLite是文件型数据库，不需要额外安装服务，用Python就能直接操作：

第一步：把CSV导入SQLite #

import sqlite3
import csv

# 连接到数据库文件（不存在会自动创建）
conn = sqlite3.connect('large_file_db.sqlite')
cursor = conn.cursor()

# 读取CSV表头，创建表结构
with open('your_large_file.csv', 'r', newline='') as csv_file:
    reader = csv.reader(csv_file)
    headers = next(reader)
    # 生成建表语句，默认所有列设为文本类型，你可以根据实际数据调整
    create_table_sql = f"CREATE TABLE IF NOT EXISTS data ({', '.join([f'{col} TEXT' for col in headers])})"
    cursor.execute(create_table_sql)
    
    # 分块插入数据，避免内存溢出
    chunk_size = 10000
    chunk = []
    for row in reader:
        chunk.append(tuple(row))
        if len(chunk) == chunk_size:
            cursor.executemany(f"INSERT INTO data VALUES ({', '.join(['?']*len(headers))})", chunk)
            chunk = []
    # 插入剩余的数据
    if chunk:
        cursor.executemany(f"INSERT INTO data VALUES ({', '.join(['?']*len(headers))})", chunk)
    
    conn.commit()

# 关闭连接
conn.close()

第二步：查询特定行 #

之后你就可以用SQL语句精准查询需要的行，比如按行号（SQLite自带rowid字段，从1开始）：

import sqlite3

target_row_ids = [101, 2001, 50001]  # 对应原CSV的第100、2000、50000行数据（因为rowid从1开始，表头是rowid=1）
conn = sqlite3.connect('large_file_db.sqlite')
cursor = conn.cursor()

# 执行查询
cursor.execute(f"SELECT * FROM data WHERE rowid IN ({', '.join(['?']*len(target_row_ids))})", target_row_ids)
selected_rows = cursor.fetchall()

# 关闭连接
conn.close()

HDF5格式（传统大文件存储） #

HDF5也是常用的大文件存储格式，支持随机访问，Pandas原生支持：

转格式并读取 #

import pandas as pd

# 转成HDF5（同样可以分块写入）
with pd.HDFStore('your_large_file.h5') as store:
    store['data'] = pd.read_csv('your_large_file.csv')

# 读取特定行
with pd.HDFStore('your_large_file.h5') as store:
    selected_df = store['data'].iloc[[100, 2000, 50000]]

不过HDF5的压缩率和跨平台性不如Parquet，现在更推荐用Parquet。

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