如何逐行读取大文件？

问题描述：

我想遍历整个文件的每一行。一种方法是读取整个文件，将其保存到列表中，然后遍历感兴趣的行。这种方法占用大量内存，所以我正在寻找替代方法。

我迄今为止的代码：

for each_line in fileinput.input(input_file):
    do_something(each_line)

    for each_line_again in fileinput.input(input_file):
        do_something(each_line_again)

执行此代码会给出错误消息：device active。

有什么建议吗？

目的是计算成对的字符串相似度，也就是说，对于文件中的每一行，我想计算与其他每一行的 Levenshtein 距离。

2022 年 11 月编辑：这个问题提出 8 个月后提出的一个相关问题有很多有用的答案和评论。为了更深入地了解 Python 逻辑，也请阅读这个相关问题我应该如何在 Python 中逐行读取文件？

解决方案 1：

读取文件的正确且完全 Pythonic 的方式如下：

with open(...) as f:
    for line in f:
        # Do something with 'line'

该with语句处理文件的打开和关闭，包括内部块中是否出现异常。它将for line in f文件对象f视为可迭代对象，自动使用缓冲 I/O 和内存管理，因此您不必担心大文件。

应该有一个（最好只有一个）明显的方法来做到这一点。

解决方案 2：

按排序顺序排列的两种内存高效方法（第一种是最好的）-

1. 使用with- 从 Python 2.5 及以上版本开始支持

2. yield如果你真的想控制阅读量，可以使用

3. 使用with

with是读取大文件的一种很好的、​​高效的 Python 方式。优点 - 1）退出with执行块后文件对象自动关闭。2）with块内的异常处理。3）内存for循环逐行遍历f文件对象。内部它执行缓冲 IO（对昂贵的 IO 操作进行优化）和内存管理。

with open("x.txt") as f:
    for line in f:
        do something with data

1. 使用yield

有时，人们可能希望对每次迭代中读取的内容进行更细粒度的控制。在这种情况下，请使用iter & Yield。请注意，使用此方法时，需要在最后关闭文件。

def readInChunks(fileObj, chunkSize=2048):
    """
    Lazy function to read a file piece by piece.
    Default chunk size: 2kB.

    """
    while True:
        data = fileObj.read(chunkSize)
        if not data:
            break
        yield data

f = open('bigFile')
for chunk in readInChunks(f):
    do_something(chunk)
f.close()

陷阱和为了完整性- 下面的方法对于读取大文件来说并不那么好或不那么优雅，但请阅读以获得全面的理解。

在 Python 中，从文件读取行的最常见方法是执行以下操作：

for line in open('myfile','r').readlines():
    do_something(line)

但是，完成此操作后，readlines()函数（函数同样适用read()）会将整个文件加载到内存中，然后对其进行迭代。对于大文件，一种稍微好一点的方法（上面提到的两种方法是最好的）是使用模块fileinput，如下所示：

import fileinput

for line in fileinput.input(['myfile']):
    do_something(line)

该fileinput.input()调用按顺序读取行，但读取后并不将它们保存在内存中，甚至只是这样，因为file在 python 中是可迭代的。

参考

1. Python with 语句

解决方案 3：

删除换行符：

with open(file_path, 'rU') as f:
    for line_terminated in f:
        line = line_terminated.rstrip('
')
        ...

有了通用换行符支持，所有文本文件行似乎都会以 终止`'
'，无论文件中的终止符是什么'
'，'
'、 或'
'`。

编辑-指定通用换行符支持：

• Unix 上的 Python 2——open(file_path, mode='rU')必需[感谢@Dave ]

• Windows 上的 Python 2——open(file_path, mode='rU')可选

• Python 3——open(file_path, newline=None)可选

参数newline仅在 Python 3 中受支持，默认为None，表示通用换行符模式（输入文件可以有任何换行符，输出字符串为`
）。在所有情况下，mode参数默认为。 Python 3 中已弃用 。 在 Windows 上的 Python 2 中，似乎有其他一些机制可以转换为。'r'U
``
`

文档：

• Python 2 中的 open()

• Python 3 中的 open()

要保留本机行终止符：

with open(file_path, 'rb') as f:
    with line_native_terminated in f:
        ...

二进制模式仍然可以使用 将文件解析为行in。每行将具有文件中存在的终止符。

感谢@katrielalex的回答、Python 的open()文档和iPython实验。

解决方案 4：

这是在 python 中读取文件的一种可能方法：

f = open(input_file)
for line in f:
    do_stuff(line)
f.close()

它不分配完整的列表。它迭代所有行。

解决方案 5：

先介绍一下我的观点。代码片段在最后。

如果可以，我更喜欢使用开源工具（如 H2O）来执行超高性能并行 CSV 文件读取，但此工具的功能集有限。我最终编写了大量代码来创建数据科学管道，然后才将其输入到 H2O 集群进行适当的监督学习。

我一直在读取 UCI repo 中的 8GB HIGGS 数据集等文件，甚至读取 40GB CSV 文件以用于数据科学目的，通过添加大量并行处理库的池对象和映射函数，速度显著提高。例如，使用最近邻搜索进行聚类以及 DBSCAN 和马尔可夫聚类算法需要一些并行编程技巧来绕过一些极具挑战性的内存和挂钟时间问题。

我通常喜欢先使用 gnu 工具将文件按行拆分成多个部分，然后使用 glob-filemask 将它们全部查找并并行读取到 Python 程序中。我通常使用 1000 多个部分文件。使用这些技巧对处理速度和内存限制有很大帮助。

pandas dataframe.read_csv 是单线程的，因此您可以通过运行 map() 进行并行执行来使用这些技巧，使 pandas 运行得更快。您可以使用 htop 查看，使用普通的顺序 pandas dataframe.read_csv，pd.read_csv 中的实际瓶颈是单个核心上的 100% CPU，而不是磁盘。

我应该补充一下，我在快速显卡总线上使用的是 SSD，而不是 SATA6 总线上的旋转 HD，另外还有 16 个 CPU 核心。

此外，我发现另一种在某些应用程序中效果很好的技术是并行 CSV 文件读取一个巨型文件中的所有内容，每个工作程序从文件的不同偏移量启动，而不是将一个大文件预先分割成许多部分文件。在每个并行工作程序中使用 python 的文件 seek() 和 tell() 以条带形式读取大文本文件，在大文件中的不同字节偏移起始字节和结束字节位置同时同时读取。您可以对字节执行正则表达式 findall，并返回换行符的数量。这是一个部分和。最后，当工作程序完成后 map 函数返回时，将部分和相加以获得全局和。

以下是一些使用并行字节偏移技巧的示例基准测试：

我使用了 2 个文件：HIGGS.csv 是 8 GB。它来自 UCI 机器学习存储库。all_bin .csv 是 40.4 GB，来自我当前的项目。我使用了 2 个程序：Linux 自带的 GNU wc 程序和我开发的纯 Python fastread.py 程序。

HP-Z820:/mnt/fastssd/fast_file_reader$ ls -l /mnt/fastssd/nzv/HIGGS.csv
-rw-rw-r-- 1 8035497980 Jan 24 16:00 /mnt/fastssd/nzv/HIGGS.csv

HP-Z820:/mnt/fastssd$ ls -l all_bin.csv
-rw-rw-r-- 1 40412077758 Feb  2 09:00 all_bin.csv

ga@ga-HP-Z820:/mnt/fastssd$ time python fastread.py --fileName="all_bin.csv" --numProcesses=32 --balanceFactor=2
2367496

real    0m8.920s
user    1m30.056s
sys 2m38.744s

In [1]: 40412077758. / 8.92
Out[1]: 4530501990.807175

这相当于 4.5 GB/s 或 45 Gb/s 的文件读取速度。这不是旋转硬盘，我的朋友。这实际上是三星 Pro 950 SSD。

下面是纯 C 编译程序 gnu wc 对同一文件进行行计数的速度基准测试。

很酷的是，你可以看到，在这种情况下，我的纯 Python 程序基本上与 gnu wc 编译的 C 程序的速度相当。Python 是解释型的，但 C 是编译型的，所以这是一个非常有趣的速度壮举，我想你会同意的。当然，wc 确实需要改为并行程序，然后它才会真正击败我的 Python 程序。但就目前而言，gnu wc 只是一个顺序程序。你尽你所能，python 现在就可以并行。Cython 编译也许能帮我（在其他时候）。此外，内存映射文件尚未探索。

HP-Z820:/mnt/fastssd$ time wc -l all_bin.csv
2367496 all_bin.csv

real    0m8.807s
user    0m1.168s
sys 0m7.636s


HP-Z820:/mnt/fastssd/fast_file_reader$ time python fastread.py --fileName="HIGGS.csv" --numProcesses=16 --balanceFactor=2
11000000

real    0m2.257s
user    0m12.088s
sys 0m20.512s

HP-Z820:/mnt/fastssd/fast_file_reader$ time wc -l HIGGS.csv
11000000 HIGGS.csv

real    0m1.820s
user    0m0.364s
sys 0m1.456s

结论：与 C 程序相比，纯 Python 程序的速度不错。但是，使用纯 Python 程序比使用 C 程序还不够好，至少对于行计数而言。通常，该技术可以用于其他文件处理，因此这个 Python 代码仍然很好。

问题：只编译一次正则表达式并将其传递给所有工作人员是否会提高速度？ 答案：正则表达式预编译对此应用程序没有帮助。我猜原因是所有工作人员的进程序列化和创建的开销占主导地位。

还有一件事。并行读取 CSV 文件有帮助吗？瓶颈是磁盘还是 CPU？许多所谓的 stackoverflow 顶级答案都包含常见的开发智慧，即只需要一个线程来读取文件，这是你能做到的最好程度。但他们确定吗？

让我们来一探究竟：

HP-Z820:/mnt/fastssd/fast_file_reader$ time python fastread.py --fileName="HIGGS.csv" --numProcesses=16 --balanceFactor=2
11000000

real    0m2.256s
user    0m10.696s
sys 0m19.952s

HP-Z820:/mnt/fastssd/fast_file_reader$ time python fastread.py --fileName="HIGGS.csv" --numProcesses=1 --balanceFactor=1
11000000

real    0m17.380s
user    0m11.124s
sys 0m6.272s

哦，是的，确实如此。并行文件读取效果很好。好了！

附言：有些人想知道，如果使用单个工作进程时 balanceFactor 为 2 会怎么样？嗯，这太可怕了：

HP-Z820:/mnt/fastssd/fast_file_reader$ time python fastread.py --fileName="HIGGS.csv" --numProcesses=1 --balanceFactor=2
11000000

real    1m37.077s
user    0m12.432s
sys 1m24.700s

fastread.py python 程序的关键部分：

fileBytes = stat(fileName).st_size  # Read quickly from OS how many bytes are in a text file
startByte, endByte = PartitionDataToWorkers(workers=numProcesses, items=fileBytes, balanceFactor=balanceFactor)
p = Pool(numProcesses)
partialSum = p.starmap(ReadFileSegment, zip(startByte, endByte, repeat(fileName))) # startByte is already a list. fileName is made into a same-length list of duplicates values.
globalSum = sum(partialSum)
print(globalSum)


def ReadFileSegment(startByte, endByte, fileName, searchChar='
'):  # counts number of searchChar appearing in the byte range
    with open(fileName, 'r') as f:
        f.seek(startByte-1)  # seek is initially at byte 0 and then moves forward the specified amount, so seek(5) points at the 6th byte.
        bytes = f.read(endByte - startByte + 1)
        cnt = len(re.findall(searchChar, bytes)) # findall with implicit compiling runs just as fast here as re.compile once + re.finditer many times.
    return cnt

PartitionDataToWorkers 的定义只是普通的顺序代码。我省略了它，以防其他人想要练习一下并行编程是什么样的。我免费提供了较难的部分：经过测试和工作的并行代码，供您学习。

感谢：Arno 和 Cliff 的开源 H2O 项目以及 H2O 员工提供的出色软件和教学视频，为我提供了灵感，让我开发出如上所示的纯 Python 高性能并行字节偏移读取器。H2O 使用 Java 进行并行文件读取，可由 Python 和 R 程序调用，速度非常快，比世界上任何读取大型 CSV 文件的方法都快。

解决方案 6：

Katrielalex 提供了打开和读取一个文件的方法。

但是，您的算法会读取文件的每一行。这意味着，如果 N 是文件中的行数，则读取文件的总次数（以及计算Levenshtein 距离）将为 N*N。由于您关心文件大小并且不想将其保存在内存中，因此我关心的是由此产生的二次运行时间。您的算法属于 O(n^2) 类算法，通常可以通过专业化进行改进。

我猜你已经知道了内存与运行时间之间的权衡，但也许你想研究一下是否有一种高效的方法来并行计算多个编辑距离。如果是这样的话，在这里分享你的解决方案会很有趣。

你的文件有多少行，你的算法必须在什么样的机器（内存和 CPU 能力）上运行，以及可容忍的运行时间是多少？

代码如下：

with f_outer as open(input_file, 'r'):
    for line_outer in f_outer:
        with f_inner as open(input_file, 'r'):
            for line_inner in f_inner:
                compute_distance(line_outer, line_inner)

但问题是如何存储距离（矩阵？）以及您是否可以获得准备例如 outer_line 进行处理或缓存一些中间结果以供重用的优势。

解决方案 7：

需要频繁地从最后一个位置读取大文件吗？

我创建了一个脚本，用于每天多次剪切 Apache access.log 文件。因此，我需要将光标定位在上次执行期间解析的最后一行上。为此，我使用了file.seek()和file.seek()方法，这些方法允许将光标存储在文件中。

我的代码：

ENCODING = "utf8"
CURRENT_FILE_DIR = os.path.dirname(os.path.abspath(__file__))

# This file is used to store the last cursor position
cursor_position = os.path.join(CURRENT_FILE_DIR, "access_cursor_position.log")

# Log file with new lines
log_file_to_cut = os.path.join(CURRENT_FILE_DIR, "access.log")
cut_file = os.path.join(CURRENT_FILE_DIR, "cut_access", "cut.log")

# Set in from_line 
from_position = 0
try:
    with open(cursor_position, "r", encoding=ENCODING) as f:
        from_position = int(f.read())
except Exception as e:
    pass

# We read log_file_to_cut to put new lines in cut_file
with open(log_file_to_cut, "r", encoding=ENCODING) as f:
    with open(cut_file, "w", encoding=ENCODING) as fw:
        # We set cursor to the last position used (during last run of script)
        f.seek(from_position)
        for line in f:
            fw.write("%s" % (line))

    # We save the last position of cursor for next usage
    with open(cursor_position, "w", encoding=ENCODING) as fw:
        fw.write(str(f.tell()))

解决方案 8：

来自fileinput .input()的 python 文档：

这将遍历列出的所有文件的行，如果列表为空，sys.argv[1:]则默认为sys.stdin

进一步，该函数的定义是：

fileinput.FileInput([files[, inplace[, backup[, mode[, openhook]]]]])

从字里行间可以看出，这files可以是一个列表，因此您可以得到类似以下内容的内容：

for each_line in fileinput.input([input_file, input_file]):
  do_something(each_line)

请参阅此处了解更多信息

解决方案 9：

我强烈建议不要使用默认文件加载，因为它太慢了。你应该研究一下 numpy 函数和 IOpro 函数（例如 numpy.loadtxt()）。

http://docs.scipy.org/doc/numpy/user/basics.io.genfromtxt.html

https://store.continuum.io/cshop/iopro/

然后，你可以将成对操作分解成块：

import numpy as np
import math

lines_total = n    
similarity = np.zeros(n,n)
lines_per_chunk = m
n_chunks = math.ceil(float(n)/m)
for i in xrange(n_chunks):
    for j in xrange(n_chunks):
        chunk_i = (function of your choice to read lines i*lines_per_chunk to (i+1)*lines_per_chunk)
        chunk_j = (function of your choice to read lines j*lines_per_chunk to (j+1)*lines_per_chunk)
        similarity[i*lines_per_chunk:(i+1)*lines_per_chunk,
                   j*lines_per_chunk:(j+1)*lines_per_chunk] = fast_operation(chunk_i, chunk_j) 

分块加载数据然后对其执行矩阵运算几乎总是比逐个元素地执行操作快得多！

解决方案 10：

#Using a text file for the example
with open("yourFile.txt","r") as f:
    text = f.readlines()
for line in text:
    print line

• 打开文件进行阅读 (r)

• 读取整个文件并将每一行保存到列表（文本）中

• 循环遍历列表并打印每一行。

例如，如果您想要检查长度大于 10 的特定行，请使用现有资源。

for line in text:
    if len(line) > 10:
        print line

解决方案 11：

逐行读取大文件的最佳方法是使用 python枚举函数

with open(file_name, "rU") as read_file:
    for i, row in enumerate(read_file, 1):
        #do something
        #i in line of that line
        #row containts all data of that line