python - 如何有效地将带有 BZ2 JSON twitter 文件的大型 (30GB+) TAR 文件读入 PostgreSQL

Question

我正在尝试从 archive.org archive 获取推特数据并将其加载到数据库中。我试图首先加载特定月份的所有 推文，然后选择推文并仅展示我感兴趣的推文(例如按区域设置或主题标签)。

我能够运行下面描述的脚本来完成我正在寻找的事情，但我有一个问题，它非常慢。它运行了大约半小时，并且只读取了一个 TAR 文件中的 ~ 6/50,000 个内部 .bz2 文件。

示例 TAR 文件的一些统计信息:

• 总大小:~ 30-40GB
• 内部 .bz2 文件数量(按文件夹排列):50,000
• 一个 .bz2 文件的大小:~600kb
• 一个提取的 JSON 文件的大小:~5 MB，~3600 条推文。

在优化此过程以提高速度时我应该注意什么？

• 我应该将文件提取到磁盘而不是用 Python 缓冲它们吗？
• 我是否应该将多线程视为流程的一部分？流程的哪一部分对此是最佳的？
• 或者，对于这样的脚本，我目前获得的速度是否相对正常？

脚本目前使用了 ~ 3% 的 CPU 和 ~ 6% 的 RAM 内存。

非常感谢任何帮助。

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 .因此，当您提取一个文件时，必须读取整个 tar 文件。即使在找到该文件之后，仍必须读取 tar 文件的其余部分以检查是否存在后续副本。

这使得提取一个文件与提取所有文件一样昂贵。

因此，切勿在大型 tar 文件上使用 tar.extractfile(...)(除非您只需要一个文件或没有足够的空间来提取所有内容)。

如果您有足够的空间(考虑到现代硬盘驱动器的大小，您几乎肯定有)，请使用 tar.extractall 提取所有内容或者通过系统调用 tar xf ...，然后处理提取的文件。