巨大なテキストファイル (数 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 行のバッチを処理したいのですが、何かが足りないことです。
islice()nイテレータの次の項目を取得するために使用できます。したがって、ファイルの次の行のlist(islice(f, n))リストを返します。これをループ内で使用すると、ファイルが行のチャンクで表示されます。ファイルの最後では、リストが短くなる可能性があり、最終的に呼び出しは空のリストを返します。nfn
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
質問は、一度に N 行のブロックで「巨大なテキストファイル」を読み取ることによって効率が得られることを前提としているようです。これにより、すでに高度に最適化されたstdioライブラリにバッファリングのアプリケーション層が追加され、複雑さが増し、おそらく何も得られません。
したがって：
with open('my_very_large_text_file') as f:
for line in f:
process(line)
おそらく、時間、スペース、複雑さ、読みやすさの点で、どの代替手段よりも優れています。
Rob Pike の最初の 2 つの規則、Jackson の 2 つの規則、およびPEP-20 The Zen of Pythonも参照してください。本当にただ遊びたいだけisliceなら、大きなファイルのものは除外すべきです。
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（）はキーパラメータの戻り値で行をグループ化し、キーパラメータはラムダ関数であり、関数が定義されるときにc引数がcount（）lambda _, c=count(): c.next()/Nにバインドされるという事実を使用して、毎回呼び出すラムダ関数と戻り値を評価して、行をグループ化するグルーパーを決定します。groupby()
# 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
...
ファイルから選択された行が統計的に均一に分布しているという要件が追加されたため、この単純なアプローチを提供します。
"""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;のさまざまな値について 100のときnsamp、シングルsample_linesは460µsで完了し、呼び出しごとに47msで最大10kサンプルまで線形にスケーリングされました。
次の自然な質問は、ランダムはほとんどランダムではないということです。、そして答えは「サブ暗号ですが、バイオインフォマティクスには確かに問題ありません」です。
チャンクのリストを反復処理するための最も「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]"""
「バッチ」とは、16 個のレコードすべてを個別ではなく一度に処理することを意味し、一度に 1 レコードずつファイルを読み取り、カウンターを更新することを意味します。カウンターが 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 = 0the final group
process_list(interim_list)