在python中将数据块加载到.json（或字典）中

Question

我想通过请求流式下载一个 .gz 文件并同时解压缩流式文件。

我想出的代码位是：

import requests
import zlib

url = "https://something"
d = zlib.decompressobj(zlib.MAX_WBITS | 16)
with requests.get(url, stream=True) as r:
    r.raise_for_status()
    for chunk in r.iter_content(chunk_size=128):  
        data = d.decompress(chunk)

我还想将未压缩的数据块转换为 json/ 字典。但是，由于不能保证数据的分块部分是完整的 json/字典格式，我无法将数据加载到字典或 json 格式中。然后，我想将数据位发送到将数据插入 mongodb 数据库的管道。

我想做这样的事情：

import requests
import json
import pandas as pd
import zlib

url = "https://something"
d = zlib.decompressobj(zlib.MAX_WBITS | 16)
with requests.get(url, stream=True) as r:
    r.raise_for_status()
    for chunk in r.iter_content(chunk_size=128):  
        data = d.decompress(chunk)

        #something like the line down there
        yield json.load(data)

        #or something like 
        yield df.to_dict(data)

是否有将这些未压缩的块加载到 json 或将它们转换为字典？或者，以其他方式将此数据加载到 mongodb 中？

仅供参考，我的压缩文件约为 90MB，未压缩文件约为 1.2 GB。

Answer 1

我找到了一个解决我的问题的方法，它对我有用，但不确定它是否适用于其他任何人。我想我会在这里分享它以防万一。

import json
import requests
import zlib

url = "helloworld.com"
data = ""
flag = 0
d = zlib.decompressobj(zlib.MAX_WBITS | 16)
with requests.get(url, stream=True) as r:
    r.raise_for_status()
    for chunk in r.iter_content(chunk_size=128):  
        datum = d.decompress(chunk)
        try:
                data+= datum.decode('utf-8')

                if flag == 1:
                        data = data[data.find('"first_key"'):]
                        data = "[{" + data
                        flag = 0



                if data.count('"first_key"')>=2:
                #here -8 to get the last instance of the first_key of an element, might change depending upon the format
                        location = data.rfind('"first_key"')-8 
                        data1 = data[:location]
                        if data1[0] == "[":
                                data1+="]"
                        data2 = json.loads(data1)
                        data = "[{" + data[data.rfind('"first_key"'):]
                        yield data2
        except:
                data = ""
                flag = 1
                continue

Answer 2

我的压缩文件大约 90MB，未压缩文件大约 1.2GB

如果你被允许使用外部模块，我建议看看json-stream 外部模块，因为

与json.load() 不同，json-stream 可以流式传输 JSON 数据来自 类似文件的对象。这有以下好处：

• 不需要将整个 json 文档预先存入内存
• 它可以在整个文档加载完成之前开始生成数据
• 它只需要足够的内存来保存当前正在解析的数据

json_stream.loads 除外类似文件的对象，如果要从内存中加载，则需要使用 io.StringIO，请考虑以下示例

import io
import json_stream
data = '["uno","dos","tres"]'
data_io = io.StringIO(data)
for i in json_stream.load(data_io):
    print(i)

输出

uno
dos
tres

请注意，io.BytesIO 和 io.StringIO 也可能用于其他需要类似文件的地方，而不仅仅是json_stream.load。 io 是内置模块，更多信息请咨询io docs。