BZ2 JSON twitterファイルを使用して大きな（30GB +）TARファイルをPostgreSQLに効率的に読み込む方法

Question

archive.org archiveからTwitterデータを取得してデータベースにロードしようとしています。特定の月のすべてのつぶやきを読み込んで、つぶやきの選択を行い、興味のあるもののみを選択します（ロケールやハッシュタグなど）。

私が探していることをするために以下に説明するスクリプトを実行することができますが、それは信じられないほど遅いという点で問題があります。これは約30分実行され、1つのTARファイル内の.bz2ファイルのうち〜6/50,000だけを読み込みます。

例のTARファイルのいくつかの統計：

• 合計サイズ：（フォルダに配置された）内部.bz2ファイルの〜30〜40ギガバイト
• 数：1つの.bz2ファイルの50,000
• サイズ：〜600kb
• 抽出されたJSONファイルのサイズ：〜5 MB、〜3600個のつぶやき。

スピードのためにこのプロセスを最適化するにはどうすればよいですか？

• ファイルをPythonでバッファリングするのではなく、ディスクに抽出する必要がありますか？
• プロセスの一部をマルチスレッディングする必要がありますか？プロセスのどの部分がこれに最適でしょうか？
• また、私が現在取得している速度は、そのようなスクリプトに対して比較的普通ですか？

スクリプトは、現在、CPUの約3％、RAMメモリの約6％を使用しています。

ご協力いただきまして誠にありがとうございます。

import tarfile 
import dataset # Using dataset as I'm still iteratively developing the table structure(s) 
import json 
import datetime 


def scrape_tar_contents(filename): 
    """Iterates over an input TAR filename, retrieving each .bz2 container: 
     extracts & retrieves JSON contents; stores JSON contents in a postgreSQL database""" 
    tar = tarfile.open(filename, 'r') 
    inner_files = [filename for filename in tar.getnames() if filename.endswith('.bz2')] 

    num_bz2_files = len(inner_files) 
    bz2_count = 1 
    print('Starting work on file... ' + filename[-20:]) 
    for bz2_filename in inner_files: # Loop over all files in the TAR archive 
     print('Starting work on inner file... ' + bz2_filename[-20:] + ': ' + str(bz2_count) + '/' + str(num_bz2_files)) 
     t_extract = tar.extractfile(bz2_filename) 
     data = t_extract.read() 
     txt = bz2.decompress(data) 

     tweet_errors = 0 
     current_line = 1 
     num_lines = len(txt.split('\n')) 
     for line in txt.split('\n'): # Loop over the lines in the resulting text file. 
      if current_line % 100 == 0: 
       print('Working on line ' + str(current_line) + '/' + str(num_lines)) 
       try: 
        tweet = json.loads(line) 
       except ValueError, e: 
        error_log = {'Date_time': datetime.datetime.now(), 
           'File_TAR': filename, 
           'File_BZ2': bz2_filename, 
           'Line_number': current_line, 
           'Line': line, 
           'Error': str(e)} 
        tweet_errors += 1 
        db['error_log'].upsert(error_log, ['File_TAR', 'File_BZ2', 'Line_number']) 
        print('Error occured, now at ' + str(tweet_errors)) 
       try: 
        tweet_id = tweet['id'] 
        tweet_text = tweet['text'] 
        tweet_locale = tweet['lang'] 
        created_at = tweet['created_at'] 
        tweet_json = tweet 
        data = {'tweet_id': tweet_id, 
          'tweet_text': tweet_text, 
          'tweet_locale': tweet_locale, 
          'created_at_str': created_at, 
          'date_loaded': datetime.datetime.now(), 
          'tweet_json': tweet_json} 
        db['tweets'].upsert(data, ['tweet_id']) 
       except KeyError, e: 
        error_log = {'Date_time': datetime.datetime.now(), 
           'File_TAR': filename, 
           'File_BZ2': bz2_filename, 
           'Line_number': current_line, 
           'Line': line, 
           'Error': str(e)} 
        tweet_errors += 1 
        db['error_log'].upsert(error_log, ['File_TAR', 'File_BZ2', 'Line_number']) 
        print('Error occured, now at ' + str(tweet_errors)) 
        continue 

if __name__ == "__main__": 
    with open("postgresConnecString.txt", 'r') as f: 
     db_connectionstring = f.readline() 
    db = dataset.connect(db_connectionstring) 

    filename = r'H:/Twitter datastream/Sourcefiles/archiveteam-twitter-stream-2013-01.tar' 
    scrape_tar_contents(filename) 

Answer 1

ファイルの場所のインデックスがtarファイルに含まれていません。さらに、tarファイルにはmore than one copy of the same fileを含めることができます。したがって、1つのファイルを抽出するときは、のtarファイル全体をと読み替える必要があります。ファイルが見つかった後でも、後のコピーが存在するかどうかを確認するために残りのtarファイルを読み込む必要があります。

これは、すべてのファイルを抽出するのと同じくらい高価な1つのファイルの抽出を行います。

したがって、大型のtarファイルでは、tar.extractfile(...)を使用しないでください（1つのファイルだけが必要な場合や、すべてを抽出する領域がない場合を除き）。

スペースがある場合（現代のハードドライブのサイズがわかっている場合）は、tar.extractallまたはシステムコールtar xf ...ですべてを抽出し、抽出したファイルを処理します。