python - 非常に大きなファイルの最適化された処理

Question

私のタスクは比較的単純です。入力ファイルの各行について、その行が特定の条件セットを満たしているかどうかをテストし、満たしていれば、その行の特定の列を新しいファイルに書き込みます。私はこれを行う python スクリプトを作成しましたが、1) 速度の向上、2) 列名に関して作業するための最良の方法 (列番号はファイルごとに異なる可能性があるため)、および 3 について助けてください。 ) フィルター条件と目的の出力列を指定する最良の方法です。

1) 私が扱っているファイルには、天体画像の測光が含まれています。各ファイルは約 1e6 行×150 列の float であり、通常、サイズは 1GB を超えます。このようなファイルを約 1 分で処理する古い AWK スクリプトがあります。私の python スクリプトには 5 ～ 7 分かかります。出力ファイルが目的のものになるまで、フィルタリング条件を微調整して数回再実行する必要があることがよくあるため、速度は間違いなく望ましいものです。for ループは非常に高速であることがわかりました。ループを遅くするのは、ループ内での処理方法です。itemgetter を使用して必要な列だけを選択することは、行全体をメモリに読み込むよりも大幅に改善されましたが、さらに速度を上げるために何ができるかわかりません。これは AWK と同じくらい高速ですか?

2) 特定の量 (フォトン数、バックグラウンド、シグナル対ノイズなど) の列番号はファイル間で変わる可能性があるため、列番号ではなく列名で作業したいと思います。私の AWK スクリプトでは、フィルタリングと出力が同じ量に適用される場合でも、条件と出力列が指定されている場所で列番号が正しいことを常に確認する必要があります。Python での私の解決策は、各数量に列番号を割り当てる辞書を作成することでした。ファイルに異なる列がある場合、新しい辞書を指定するだけで済みます。おそらくこれを行うためのより良い方法がありますか？

3) 理想的には、入力ファイルと出力ファイルの名前、フィルタリング条件、および出力する目的の列を指定するだけでよく、それらはスクリプトの先頭にあるため、検索する必要はありません。何かを微調整するためだけのコード。私の主な問題は、未定義の変数にありました。たとえば、一般的な条件は「SNR > 4」ですが、「SNR」(信号対ノイズ) には、測光ファイルからラインの読み取りが開始されるまで、実際には値が割り当てられません。私の解決策は、文字列と eval/exec の組み合わせを使用することでした。繰り返しますが、もっと良い方法があるのではないでしょうか?

私はコンピューター サイエンスの訓練を受けていません (私は天文学の大学院生です)。通常、何かを一緒にハックして、それが機能するまでデバッグします。しかし、上記の 3 点に関する最適化は、私の研究にとって非常に重要になってきました。長文で申し訳ありませんが、詳しく教えていただけると助かります。クリーンアップ/コーディングスタイルに加えて、あなたが私に持っているすべてのアドバイスをいただければ幸いです。

どうもありがとう、ジェイク

#! /usr/bin/env python2.6

from operator import itemgetter


infile = 'ugc4305_1.phot'
outfile = 'ugc4305_1_filt.phot'

# names must belong to dicitonary
conditions = 'OBJ <= 2 and SNR1 > 4 and SNR2 > 4 and FLAG1 < 8 and FLAG2 < 8 and (SHARP1 + SHARP2)**2 < 0.075 and (CROWD1 + CROWD2) < 0.1'

input = 'OBJ, SNR1, SNR2, FLAG1, FLAG2, SHARP1, SHARP2, CROWD1, CROWD2'
    # should contain all quantities used in conditions

output = 'X, Y, OBJ, COUNTS1, BG1, ACS1, ERR1, CHI1, SNR1, SHARP1, ROUND1, CROWD1, FLAG1, COUNTS2, BG2, ACS2, ERR2, CHI2, SNR2, SHARP2, ROUND2, CROWD2, FLAG2'

# dictionary of col. numbers for the more important qunatities
columns = dict(EXT=0, CHIP=1, X=2, Y=3, CHI_GL=4, SNR_GL=5, SHARP_GL=6, ROUND_GL=7, MAJAX_GL=8, CROWD_GL=9, OBJ=10, COUNTS1=11, BG1=12, ACS1=13, STD1=14, ERR1=15, CHI1=16, SNR1=17, SHARP1=18, ROUND1=19, CROWD1=20, FWHM1=21, ELLIP1=22, PSFA1=23, PSFB1=24, PSFC1=25, FLAG1=26, COUNTS2=27, BG2=28, ACS2=29, STD2=30, ERR2=31, CHI2=32, SNR2=33, SHARP2=34, ROUND2=35, CROWD2=36, FWHM2=37, ELLIP2=38, PSFA2=39, PSFB2=40, PSFC2=41, FLAG2=42)



f = open(infile)
g = open(outfile, 'w')


# make string that extracts values for testing
input_items = []
for i in input.replace(',', ' ').split():
    input_items.append(columns[i])
input_items = ', '.join(str(i) for i in input_items)

var_assign = '%s = [eval(i) for i in itemgetter(%s)(line.split())]' % (input, input_items) 


# make string that specifies values for writing
output_items = []
for i in output.replace(',', ' ').split():
    output_items.append(columns[i])
output_items = ', '.join(str(i) for i in output_items)

output_values = 'itemgetter(%s)(line.split())' % output_items


# make string that specifies format for writing
string_format = []
for i in output.replace(',', ' ').split():
    string_format.append('%s')
string_format = ' '.join(string_format)+'\n'


# main loop
for line in f:
   exec(var_assign)
   if eval(conditions):
      g.write(string_format % tuple(eval(output_values)))
f.close()
g.close()

Answer 1

これが私がこのようなことをする方法です...

これは、私のマシンの元のスクリプトの場合は約3分であるのに対し、約35秒で実行されます。さらにいくつかの最適化を追加することは可能ですが（たとえば、いくつかの列をフロートに変換するだけで済みます）、実行時間の数秒を短縮するだけです。

csv.DictReader何人かの人が示唆しているように、ここでも簡単に使用できます。カスタム方言を定義する必要があるので、私はそれを避けています、そしてそれなしで同じことをするのはほんの数行の余分な行です。（さまざまなcsvモジュールクラスは、この特定のケースでは心配する必要のない、より複雑な動作（引用符で囲まれた文字列など）もチェックします。多くの場合、非常に便利ですが、この場合。）

スクリプトを呼び出すときに、ハードコーディングする代わりに、インファイル名とアウトファイル名を引数として簡単に追加することもできることに注意してください（つまりinfile = sys.argv[0]、など）。これにより、データを簡単にパイプインまたはパイプアウトすることもできます...（長さを確認して設定したり、にsys.argv設定したり、それに応じて）infileoutfilesys.stdinsys.stdout

def main():
    infile = 'ugc4305_1.phot'
    outfile = 'ugc4305_1_filt.phot'
    process_data(infile, outfile)

def filter_conditions(row):
    for key, value in row.iteritems():
        row[key] = float(value)

    cond = (row['OBJ'] <= 2 and row['SNR1'] > 4 
       and row['SNR2'] > 4 and row['FLAG1'] < 8 
       and row['FLAG2'] < 8 
       and (row['SHARP1'] + row['SHARP2'])**2 < 0.075 
       and (row['CROWD1'] + row['CROWD2']) < 0.1
       )
    return cond

def format_output(row):
    output_columns = ('X', 'Y', 'OBJ', 'COUNTS1', 'BG1', 'ACS1', 'ERR1', 'CHI1', 
                     'SNR1', 'SHARP1', 'ROUND1', 'CROWD1', 'FLAG1', 'COUNTS2', 
                     'BG2', 'ACS2', 'ERR2', 'CHI2', 'SNR2', 'SHARP2', 'ROUND2', 
                     'CROWD2', 'FLAG2')
    delimiter = '\t'
    return delimiter.join((row[name] for name in output_columns))

def process_data(infilename, outfilename):
    column_names = ('EXT', 'CHIP', 'X', 'Y', 'CHI_GL', 'SNR_GL', 'SHARP_GL', 
                    'ROUND_GL', 'MAJAX_GL', 'CROWD_GL', 'OBJ', 'COUNTS1', 
                    'BG1', 'ACS1', 'STD1', 'ERR1', 'CHI1', 'SNR1', 'SHARP1', 
                    'ROUND1', 'CROWD1', 'FWHM1', 'ELLIP1', 'PSFA1', 'PSFB1', 
                    'PSFC1', 'FLAG1', 'COUNTS2', 'BG2', 'ACS2', 'STD2', 
                    'ERR2', 'CHI2', 'SNR2', 'SHARP2', 'ROUND2', 'CROWD2', 
                    'FWHM2', 'ELLIP2', 'PSFA2', 'PSFB2', 'PSFC2', 'FLAG2')

    with open(infilename) as infile:
        with open(outfilename, 'w') as outfile:
            for line in infile:
                line = line.strip().split()
                row = dict(zip(column_names, line))
                if filter_conditions(row.copy()):
                    outfile.write(format_output(row) + '\n')

if __name__ == '__main__':
    main()

Answer 2

あなたが言及したとは思いませんが、あなたのデータは csv にあるようです。csv.DictReaderを使用すると、多くのことが得られる場合があります。一度に 1 行ずつファイルを反復処理し (すべてをメモリにロードすることを回避します)、名前で列を参照できます。

まだお持ちでない場合は、Python のプロファイラーであるcProfileもご覧ください。プログラムのどの部分が実行に最も時間がかかっているかがわかります。

Answer 3

exec()ここでの私の最初のステップは、 andeval()呼び出しを取り除くことです。文字列を評価するたびに、コンパイルしてから実行する必要があり、ファイルのすべての行で関数呼び出しのオーバーヘッドが追加されます。言うまでもなく、evalコードが乱雑でデバッグが困難になる傾向があるため、一般的には避ける必要があります。

ロジックを小さくて理解しやすい関数に入れることで、リファクタリングを開始できます。たとえば、次eval(conditions)のような関数に置き換えることができます。

def conditions(d):
    return (d[OBJ] <= 2 and
            d[SNRI] > 4 and
            d[SNR2] > 4 and
            d[FLAG1] < 8 and ...

ヒント: いくつかの条件が失敗する可能性が高い場合は、それらを最初に配置すると、python は残りの評価をスキップします。

列名の辞書を取り除き、ファイルの先頭に一連の変数を設定してから、line[COLNAME]. これにより、条件関数などの一部を簡素化でき、各変数を割り当てることなく名前で列を参照できます。

Answer 4

nmichaels が言ったように、fieldnamesとdialectパラメータを使用してcsv.DictReaderこのファイルを読み取ることができます。次に、各行に辞書があります。辞書を使用するevalと、を使用する必要がなくなり、次のようなステートメントを使用できます

if data_item['OBJ'] <= 2 and data_item['SNR1']:
     g.write(data_item['X'], data_item['Y'], data_item['OBJ'])

あなたが現在行っている方法は、すべてのevals のために遅くて複雑です。その複雑さは必要ありません。

Answer 5

ファイルの実際の読み取りと解析に多くの時間が費やされていることがプロファイリングで示され、同じ生ファイルを何度も処理する場合は、Python での読み取り用に最適化された中間ファイル形式の作成を試みることができます。

ファイルを 1 回読み取り、pickle/cPickleで結果を解析して出力することもできます。次に、フィルタ スクリプトで pickle/cpickle を使用して中間ファイルを読み取ります。

これが各行を読んでそれらを分割するよりも速いかどうかを判断するのに十分なほどpythonを知りません。(c# ではバイナリ シリアライザーを使用しますが、それが Python で利用できるかどうかはわかりません)。

ディスク IO がボトルネックである場合は、入力ファイルを圧縮してgzipモジュールで読み取ることもできます。