读取许多 CSV 最后一行的高性能方法？

在 Python 中，我想循环遍历 1000 多个包含 100k 行的 CSV 文件的列表，并且只读取其中的最后 100 行左右。将它们加载到 pandas 数据框中需要太多的时间和内存，因为它将整个文件加载到数据框中。在 Python 中执行此操作的快速有效方法是什么？

import os
import pandas as pd

directory = "/tmp/path/to/csv/"
out = []

for filename in os.listdir(directory):
    file_path = os.path.join(directory, filename)
    df = pd.read_csv(file_path)
    if np.all( df.iloc[-1] < 100):
        out.append(filename)

以下是几种在 Python 中读取 CSV 文件最后几行的高效方法：

1. 使用 tail 命令与 csv 库

这可能是最简单且通常最快的方法，特别是对于大型文件，因为它使用 Unix 实用程序 tail 来处理文件处理：

import os
import csv
import subprocess

def read_last_n_lines(file_path, n):
    """使用 tail 命令读取文件的最后 n 行。"""
    try:
        # 使用 tail 命令获取最后 n 行
        result = subprocess.run(['tail', f'-n{n}', file_path], stdout=subprocess.PIPE, text=True, check=True)
        lines = result.stdout.splitlines()
        reader = csv.reader(lines)
        return list(reader)
    except FileNotFoundError:
        print(f"文件未找到：{file_path}")
        return []

directory = "/tmp/path/to/csv/"
out = []

for filename in os.listdir(directory):
    file_path = os.path.join(directory, filename)
    last_100_rows = read_last_n_lines(file_path, 100)

    # 处理 last_100_rows 中的数据，例如：
    if last_100_rows:
        last_row = last_100_rows[-1]
        # 检查 last_row 中的值是否都小于 100
        if all(float(x) < 100 for x in last_row):
            out.append(filename) 

2. 使用文件迭代器

这种方法仍然非常高效，因为它不会将整个文件读入内存。对于更大的文件，它比 pandas 快，并且更易于理解比 deque ：

import os

def read_last_n_lines(file_path, n):
    """使用文件迭代器读取文件的最后 n 行。"""
    with open(file_path, 'rb') as f:
        # 从文件末尾开始查找换行符，以处理不同的行尾
        f.seek(-2, os.SEEK_END)
        lines_to_read = n
        line_count = 0
        while lines_to_read > 0 and f.tell() > 0:
            f.seek(-2, os.SEEK_CUR)
            if f.read(1) == b'\n':
                lines_to_read -= 1
                line_count += 1
            f.seek(-1, os.SEEK_CUR)
        return f.readlines()[-line_count:]

directory = "/tmp/path/to/csv/"
out = []

for filename in os.listdir(directory):
    file_path = os.path.join(directory, filename)
    last_100_rows = read_last_n_lines(file_path, 100)

    # 使用 csv.reader 处理字节，并检查最后一行的值
    if last_100_rows:
        import csv
        last_row = next(csv.reader([line.decode('utf-8') for line in last_100_rows[-1:]]))
        if all(float(x) < 100 for x in last_row):
            out.append(filename)

3. 使用 collections.deque

这种方法对于较小的文件非常有效，但对于非常大的文件， tail 方法可能会更快：

import os
from collections import deque

def read_last_n_lines(file_path, n):
    """使用 deque 读取文件的最后 n 行。"""
    with open(file_path, 'r') as f:
        return list(deque(f, maxlen=n))

directory = "/tmp/path/to/csv/"
out = []

for filename in os.listdir(directory):
    file_path = os.path.join(directory, filename)
    last_100_rows = read_last_n_lines(file_path, 100)

    # 处理 last_100_rows 中的数据，例如：
    if last_100_rows:
        last_row = last_100_rows[-1].split(',')
        # 检查 last_row 中的值是否都小于 100
        if all(float(x) < 100 for x in last_row):
            out.append(filename)

这些方法都避免了将整个 CSV 文件读入内存，这将提高性能，特别是当处理大量大型文件时。选择最适合的方法取决于的特定需求和文件大小。