反序列化消息而不将整个文件加载到内存中？答案

【问题标题】：Deserializing messages without loading entire file into memory?反序列化消息而不将整个文件加载到内存中？
【发布时间】：2020-02-13 13:25:34
【问题描述】：

我正在使用 Google Protocol Buffers 和 Python 来解码一些大型数据文件——每个文件 200MB。我下面有一些代码显示了如何解码分隔流，它工作得很好。但是，它使用read() 命令将整个文件加载到内存中，然后对其进行迭代。

import feed_pb2 as sfeed
import sys
from google.protobuf.internal.encoder import _VarintBytes
from google.protobuf.internal.decoder import _DecodeVarint32

with open('/home/working/data/feed.pb', 'rb') as f:
    buf = f.read() ## PROBLEM-LOADS ENTIRE FILE TO MEMORY.
    n = 0
    while n < len(buf):
        msg_len, new_pos = _DecodeVarint32(buf, n)
        n = new_pos
        msg_buf = buf[n:n+msg_len]
        n += msg_len
        read_row = sfeed.standard_feed()
        read_row.ParseFromString(msg_buf)
        # do something with read_metric
        print(read_row)

请注意，此代码来自另一个 SO 帖子，但我不记得确切的网址。我想知道是否有与协议缓冲区等效的readlines() 允许我一次读取一个分隔消息并对其进行解码？我基本上想要一个不受 RAM 限制的管道，我必须加载文件。

似乎有一个pystream-protobuf 包支持其中的一些功能，但它在一两年内没有更新。 7年前也有一篇帖子问过类似的问题。但我想知道从那以后是否有任何新信息。

python example for reading multiple protobuf messages from a stream

【问题讨论】：

我认为 SO 问题可能是这个问题：stackoverflow.com/questions/11484700/…

标签： python protocol-buffers

【解决方案1】：

如果一次加载一条完整的消息是可以的，这很容易通过修改您发布的代码来实现：

import feed_pb2 as sfeed
import sys
from google.protobuf.internal.encoder import _VarintBytes
from google.protobuf.internal.decoder import _DecodeVarint32

with open('/home/working/data/feed.pb', 'rb') as f:
    buf = f.read(10) # Maximum length of length prefix
    while buf:
        msg_len, new_pos = _DecodeVarint32(buf, 0)
        buf = buf[new_pos:]
        # read rest of the message
        buf += f.read(msg_len - len(buf))
        read_row = sfeed.standard_feed()
        read_row.ParseFromString(buf)
        buf = buf[msg_len:]
        # do something with read_metric
        print(read_row)
        # read length prefix for next message
        buf += f.read(10 - len(buf))

这会读取 10 个字节，足以解析长度前缀，然后在知道长度后读取消息的其余部分。

字符串突变在 Python 中效率不高（它们会复制大量数据），因此如果您的单个消息也很大，使用 bytearray 可以提高性能。

【讨论】：

优秀@jpa。这很有帮助。我只是希望了解这些选项，以防我的数据管道变得比我预期的要大。所以这真的很有用。我会按照你的建议调查bytearrays。
这很好，除了它仍然将整个文件读入内存（除非你在找到你想要的文件后退出）。如果您知道（或可以估计）消息的最大大小，您可以使用collections.dequeue。请参阅 this answer 以了解其用法的一个很好的示例。
NameError：未定义名称“n”。我相信代码中有错字。 n, 10 是多少？
@Riccardo 我认为是 0。

【解决方案2】：

https://github.com/cartoonist/pystream-protobuf/ 于 6 个月前更新。到目前为止，我还没有对其进行太多测试，但它似乎可以正常工作，无需任何更新。它提供可选的 gzip 和 async。

【讨论】：