我正在编写一个代码，一次取一个巨大的文本文件(几GB)N行，处理该批处理，然后移到下一行N行，直到我完成整个文件。 (我不在乎最后一批是不是完美的尺寸)。

我一直在阅读有关使用itertools islice进行此操作的信息。 我想我在那里：

from itertools import islice

N = 16

infile = open("my_very_large_text_file","r")

lines_gen = islice(infile, N)

for lines in lines_gen:

...process my lines...

麻烦的是我想处理下一批16行，但我遗漏了一些东西

可能重复使用Lazy方法在Python中读取大文件？

@ken - OP正在询问如何使用islice执行此操作，在该帖子中OP询问如何使用yield执行此操作。

可能重复如何在Python中一次读取文件N行？

islice()可用于获取迭代器的下一个n项。因此，list(islice(f, n))将返回文件f的下一个n行的列表。在循环中使用它将为您提供n行的块文件。在文件的末尾，列表可能会更短，最后调用将返回一个空列表。

from itertools import islice

with open(...) as f:

while True:

next_n_lines = list(islice(f, n))

if not next_n_lines:

break

# process next_n_lines

另一种方法是使用石斑鱼图案：

with open(...) as f:

for next_n_lines in izip_longest(*[f] * n):

# process next_n_lines

这些天我正在学习python，有一个问题，理想情况下，如果你正在阅读数据库或记录文件，你需要将记录标记为已读(需要另一列)，在下一批中你将开始处理下一个未标记的记录，这是如何实现的？特别是next_n_lines = list(islice(infile, n))

@zengr：我不明白你的问题。 list(islice(infile, n))将从文件中获取下一行n行。文件知道您已阅读的内容，您可以继续阅读。

@Sven Marnach你能评论你的代码吗？谢谢！

@Sven Say，我的批处理工作每天运行一次。我有一个1M行的巨大文本文件。但是，我只想在第1天读取前1000行。工作停止了。现在，第二天：我应该从第1001行开始处理同一个文件。那么，除了将行号计数存储在其他地方之外，你如何保持这一点。

@zengr：你必须在某个地方存放柜台。这是一个完全不相关的问题 - 使用右上角的"提问"按钮。

@Sven Marnach谢谢！您的islice代码段工作得非常棒！

我想通过nlines读取n行文件。 izip_longest()正是我所寻找的！但是，我真的不明白*[f] * n论点......任何想法？

@Stphane：这个成语在itertools.izip()的文档中提到。看看那里的等效Python代码，并记住，例如izip(*[f] * 5)相当于izip(f, f, f, f, f)。

谢谢。用test(*[f]*2)，test([f]*2)和test(f*2)做一些测试帮助我def test(*args): print args

(不可否认与原始问题不同)如果n，下一组分组的行数发生变化并在输入文件中指定，该怎么办？例如，2 a b 3 a b c =>我们想将输入分组为(a，b)(a，b，c)

@dhfromkorea：我建议使用自定义生成器功能，请参阅gist.github.com/smarnach/75146be0088e7b5c503f。

谢谢！这应该对我有用。

问题似乎是假设通过一次读取N行的块中的"巨大文本文件"可以获得效率。这为已经高度优化的stdio库增加了一个缓冲应用层，增加了复杂性，并且可能绝对没有任何东西。

从而：

with open('my_very_large_text_file') as f:

for line in f:

process(line)

在时间，空间，复杂性和可读性方面可能优于任何替代方案。

另见Rob Pike的前两条规则，Jackson的两条规则，以及PEP-20的Python之禅。如果你真的只想玩islice，你应该省去大文件的东西。

嗨！我必须在N行的块中处理我的巨大文本文件的原因是我从每组N中选择一个随机行。这是用于生物信息学分析，我想制作一个具有相同表示的较小文件。整个数据集。并非所有数据都在生物学中同等创造！可能有一种不同的(也许是更好的？)方式来选择从大量数据集中均匀分布的X个随机线，但这是我想到的第一件事。谢谢你的链接！

@brokentypewriter这是一个非常不同的问题，其中有更多具有统计学意义的样本。我会寻找现成的东西，然后把它变成一个新的问题。我会在这里放一个链接。自相关是一个令人遗憾的工件。

我在这个问题中回答了这个问题：stackoverflow.com/questions/6335839/

由于要求增加了从文件中选择的行的统计分布，我提供了这种简单的方法。

"""randsamp - extract a random subset of n lines from a large file"""

import random

def scan_linepos(path):

"""return a list of seek offsets of the beginning of each line"""

linepos = []

offset = 0

with open(path) as inf:

# WARNING: CPython 2.7 file.tell() is not accurate on file.next()

for line in inf:

linepos.append(offset)

offset += len(line)

return linepos

def sample_lines(path, linepos, nsamp):

"""return nsamp lines from path where line offsets are in linepos"""

offsets = random.sample(linepos, nsamp)

offsets.sort()  # this may make file reads more efficient

lines = []

with open(path) as inf:

for offset in offsets:

inf.seek(offset)

lines.append(inf.readline())

return lines

dataset = 'big_data.txt'

nsamp = 5

linepos = scan_linepos(dataset) # the scan only need be done once

lines = sample_lines(dataset, linepos, nsamp)

print 'selecting %d lines from a file of %d' % (nsamp, len(linepos))

print ''.join(lines)

我在磁盘上包含1.7GB的300万行的模拟数据文件上测试了它。在我不那么热门的桌面上，scan_linepos占据了运行时约20秒。

为了检查sample_lines的性能，我使用了timeit模块

import timeit

t = timeit.Timer('sample_lines(dataset, linepos, nsamp)',

'from __main__ import sample_lines, dataset, linepos, nsamp')

trials = 10 ** 4

elapsed = t.timeit(number=trials)

print u'%dk trials in %.2f seconds, %.2fμs per trial' % (trials/1000,

elapsed, (elapsed/trials) * (10 ** 6))

对于nsamp的各种值;当nsamp为100时，单个sample_lines在460μs内完成，并在每次调用47ms时线性扩展至10k个样本。

自然的下一个问题是随机几乎没有随机？，答案是"亚密码，但对生物信息学肯定很好"。

@brokentypewriter - 感谢我的实际工作o.O的愉快转移

@msw很棒的解决方案。它运行速度非常快，我喜欢random.sample采样而无需替换。唯一的问题是我在编写输出文件时出现内存错误...但我可以自己修复它。 (我将尝试的第一件事是一次写一行输出文件，而不是连接在一起的所有行)。谢谢你的解决方案！我有900万行，循环采样11次，所以节省时间的措施很棒！操作列表并将所有行加载到列表中只需要运行太长时间。

@msw我已经修复了它，每次将一行写入outfile，以避免内存问题。一切都很棒！运行需要4分25秒，这比运行上一版本(迭代列表)要好2个多小时。我真的很喜欢这个解决方案只是将从偏移中采样的行加载到内存中。这是一个简洁有效的技巧。我可以说我今天学到了新东西！

@brokentypewriter - 很高兴得到帮助，但是这种方法的功劳归功于Kernighan和Plaugher的"Pascal软件工具"(1981)，他们使用这种索引方法在没有原生字符类型的语言中实现ed(1)！一些技巧永远不会变老。

@brokentypewriter，msw：scan_linepos()不包括列表中的偏移量0，但它确实包括超过最后一行的偏移量。这意味着样本永远不会包含第一行，但如果超过最后一行的偏移量，则可能包含空行。最简单的解决方法是在for循环中交换两行。

同意，谢谢，修复。

以下是使用groupby的另一种方法：

from itertools import count, groupby

N = 16

with open('test') as f:

for g, group in groupby(f, key=lambda _, c=count(): c.next()/N):

print list(group)

这个怎么运作：

基本上groupby()将按键参数的返回值对行进行分组，键参数是lambda函数lambda _, c=count(): c.next()/N，并且使用c参数将在定义函数时绑定到count()的事实， time groupby()将调用lambda函数并计算返回值以确定将对行进行分组的分组器：

# 1 iteration.

c.next() => 0

0 / 16 => 0

# 2 iteration.

c.next() => 1

1 / 16 => 0

...

# Start of the second grouper.

c.next() => 16

16/16 => 1

...

假设"批处理"意味着要一次处理所有16个rec而不是单独处理，一次读取一个记录并更新计数器;当计数器命中16时，处理该组。

interim_list = []

infile = open("my_very_large_text_file","r")

ctr = 0

for rec in infile:

interim_list.append(rec)

ctr += 1

if ctr > 15:

process_list(interim_list)

interim_list = []

ctr = 0

the final group

process_list(interim_list)

使用chunker函数从什么是最"pythonic"的方式迭代块中的列表？：

from itertools import izip_longest

def grouper(iterable, n, fillvalue=None):

"grouper(3, 'ABCDEFG', 'x') --> ABC DEF Gxx"

args = [iter(iterable)] * n

return izip_longest(*args, fillvalue=fillvalue)

with open(filename) as f:

for lines in grouper(f, chunk_size,""): #for every chunk_sized chunk

"""process lines like

lines[0], lines[1] , ... , lines[chunk_size-1]"""

@Sven Marnach;对不起，那个"石斑鱼"必须是"chunker"。但我认为(我真的不了解你的)它与你的石斑鱼功能一样。编辑：不，不。

仍然令人困惑。 1. chunker()由两个参数定义，并用三个参数调用。 2.将f作为seq传递将尝试切片文件对象，这根本不起作用。您只能切片序列。

@Sven Marnach;实际上我首先从我的答案中得到了第一个答案，为此创建了代码，并且认为第二个答案更好，并且更改了函数，但我忘了更改函数调用。你是正确的切片，我的错误，试图纠正它。谢谢。