23

基本的に、私は次のプロセスを持っていました。

import csv
reader = csv.reader(open('huge_file.csv', 'rb'))

for line in reader:
    process_line(line)

この関連する質問を参照してください。バッチシャーディングを実装するために、100行ごとにプロセスラインを送信したいと思います。

関連する回答の実装に関する問題は、csvオブジェクトがサブスクライブできず、lenを使用できないことです。

>>> import csv
>>> reader = csv.reader(open('dataimport/tests/financial_sample.csv', 'rb'))
>>> len(reader)
Traceback (most recent call last):
  File "<stdin>", line 1, in <module>
TypeError: object of type '_csv.reader' has no len()
>>> reader[10:]
Traceback (most recent call last):
  File "<stdin>", line 1, in <module>
TypeError: '_csv.reader' object is unsubscriptable
>>> reader[10]
Traceback (most recent call last):
  File "<stdin>", line 1, in <module>
TypeError: '_csv.reader' object is unsubscriptable

どうすればこれを解決できますか？

4

3 に答える 3

28

readerにラップして、添字可能にするだけlistです。明らかに、これは非常に大きなファイルで壊れます (以下の更新プログラムの代替案を参照してください)。

>>> reader = csv.reader(open('big.csv', 'rb'))
>>> lines = list(reader)
>>> print lines[:100]
...

さらに読む: Python でリストを均等なサイズのチャンクに分割するにはどうすればよいですか?

更新 1 (リスト バージョン): 別の可能な方法は、行を反復しながら到着したときに、各チャックを処理することです。

#!/usr/bin/env python

import csv
reader = csv.reader(open('4956984.csv', 'rb'))

chunk, chunksize = [], 100

def process_chunk(chuck):
    print len(chuck)
    # do something useful ...

for i, line in enumerate(reader):
    if (i % chunksize == 0 and i > 0):
        process_chunk(chunk)
        del chunk[:]  # or: chunk = []
    chunk.append(line)

# process the remainder
process_chunk(chunk)

更新 2 (ジェネレーター バージョン): ベンチマークは行っていませんが、チャンクジェネレーターを使用してパフォーマンスを向上できる可能性があります。

#!/usr/bin/env python

import csv
reader = csv.reader(open('4956984.csv', 'rb'))

def gen_chunks(reader, chunksize=100):
    """ 
    Chunk generator. Take a CSV `reader` and yield
    `chunksize` sized slices. 
    """
    chunk = []
    for i, line in enumerate(reader):
        if (i % chunksize == 0 and i > 0):
            yield chunk
            del chunk[:]  # or: chunk = []
        chunk.append(line)
    yield chunk

for chunk in gen_chunks(reader):
    print chunk # process chunk

# test gen_chunk on some dummy sequence:
for chunk in gen_chunks(range(10), chunksize=3):
    print chunk # process chunk

# => yields
# [0, 1, 2]
# [3, 4, 5]
# [6, 7, 8]
# [9]

@totalhack が 指摘しているように、小さな落とし穴があります。

これにより、同じオブジェクトが異なる内容で何度も生成されることに注意してください。各反復の間にチャンクで必要なすべてを行うことを計画している場合、これはうまく機能します。

于 2011-02-10T12:26:11.487 に答える
2

すべてのファイルに対してこれを行う良い方法はありません。ファイルのセクションをスキップするために.csv使用して、ファイルをチャンクに分割できるはずです。file.seek次に、行の終わりを見つけるために、一度に 1 バイトずつスキャンする必要があります。2 つのチャンクを個別に処理できます。次の (テストされていない) コードのようなものから始める必要があります。

file_one = open('foo.csv')
file_two = open('foo.csv') 
file_two.seek(0, 2)     # seek to the end of the file
sz = file_two.tell()    # fetch the offset
file_two.seek(sz / 2)   # seek back to the middle
chr = ''
while chr != '\n':
    chr = file_two.read(1)
# file_two is now positioned at the start of a record
segment_one = csv.reader(file_one)
segment_two = csv.reader(file_two)

トラバースを終了したことをどのように判断できるかわかりませんsegment_one。CSV に行 ID の列があるsegment_one場合、 の最初の行の行 ID に遭遇したときに処理を停止できますsegment_two。

于 2011-02-10T12:37:38.047 に答える