在使用read（）方法从Amazon S3读取大尺寸JSON文件时出现MemoryError错误

Question

我正在尝试使用Python将一个大尺寸的JSON文件从Amazon S3导入AWS RDS PostgreSQL。但是，这些错误发生了，

回溯（最近一次通话最后一次）：

文件“my_code.py”，第67行，在

file_content=obj['Body']。read（）。解码（'utf-8'）。分割线（真）

文件"/home/user/asd-to-qwe/fgh-to-hjk/env/local/lib/python3.6/site-包/botocore/response.py"，第76行，读取

_raw_stream.read

文件“/home/user/asd to qwe/fgh to hjk/env/local/lib/python3.6/site packages/botocore/vendored/requests/packages/urllib3/response.py”，第239行，已读

数据=自我_fp。读（）

文件“/usr/lib64/python3.6/http/client.py”，第462行，已读

s=自我_安全读取（自身长度）

文件“/usr/lib64/python3.6/http/client.py”，第617行，处于“安全”状态

返回b""。加入（s）

回忆者

//我的密码。py

import sys
import boto3
import psycopg2
import zipfile
import io
import json

s3 = boto3.client('s3', aws_access_key_id=<aws_access_key_id>, aws_secret_access_key=<aws_secret_access_key>)
connection = psycopg2.connect(host=<host>, dbname=<dbname>, user=<user>, password=<password>)
cursor = connection.cursor()

bucket = sys.argv[1]
key = sys.argv[2]
obj = s3.get_object(Bucket=bucket, Key=key)

def insert_query(data):
    query = """
        INSERT INTO data_table
        SELECT
            (src.test->>'url')::varchar, (src.test->>'id')::bigint,
            (src.test->>'external_id')::bigint, (src.test->>'via')::jsonb
        FROM (SELECT CAST(%s AS JSONB) AS test) src
    """
    cursor.execute(query, (json.dumps(data),))


if key.endswith('.zip'):
    zip_files = obj['Body'].read()
    with io.BytesIO(zip_files) as zf:
        zf.seek(0)
        with zipfile.ZipFile(zf, mode='r') as z:
            for filename in z.namelist():
                with z.open(filename) as f:
                    for line in f:
                        insert_query(json.loads(line.decode('utf-8')))
if key.endswith('.json'):
    file_content = obj['Body'].read().decode('utf-8').splitlines(True)
    for line in file_content:
        insert_query(json.loads(line))


connection.commit()
connection.close()

这些问题有什么解决办法吗？任何帮助都可以，非常感谢！

Answer 1

通过避免将整个输入文件作为一行的列表写入内存，可以节省大量费用。

具体来说，这些行在内存使用方面非常糟糕，因为它们涉及到一个字节对象的峰值内存使用，即整个文件的大小，再加上一个列表，其中包含文件的完整内容：

file_content = obj['Body'].read().decode('utf-8').splitlines(True)
for line in file_content:

在64位Python 3.3上，对于一个有500万行的1GB ASCII文本文件，对于字节对象、列表和列表中的单个str而言，这是大约2.3GB的峰值内存需求。一个需要2.3倍于其处理文件大小的RAM的程序无法扩展到大文件。

要修复，请将原始代码更改为：

file_content = io.TextIOWrapper(obj['Body'], encoding='utf-8')
for line in file_content:

鉴于obj['Body']似乎可用于延迟流，这应该从内存中删除完整文件数据的两个副本。使用TextIOWrapper意味着obj['Body']被懒惰地读取和解码成块（一次几KB），并且行也被懒惰地迭代；这将内存需求降低到一个小的、基本固定的量（峰值内存开销取决于最长行的长度），而不管文件大小。

更新：

看起来StreamingBody没有实现io。缓冲数据库ABC。不过，它有自己的API文档，可以用于类似的目的。如果你不能让TextIOWrapper为你做这项工作（如果能让它工作起来，效率和简单得多），另一种选择是：

file_content = (line.decode('utf-8') for line in obj['Body'].iter_lines())
for line in file_content:

与使用TextIOWrapper不同，它不受益于块的批量解码（每行单独解码），但在减少内存使用方面，它仍然应该获得相同的好处。