Python：使列表生成器 JSON 可序列化答案

【问题标题】：Python: make a list generator JSON serializablePython：使列表生成器 JSON 可序列化
【发布时间】：2014-03-06 23:49:48
【问题描述】：

如何将 JSON 文件列表连接到一个巨大的 JSON 数组中？我有 5000 个文件和 550 000 个列表项。

我的第一次尝试是使用jq，但看起来 jq -s 没有针对大输入进行优化。

jq -s -r '[.[][]]' *.js

此命令有效，但完成时间太长，我真的很想用 Python 解决这个问题。

这是我当前的代码：

def concatFiles(outName, inFileNames):
    def listGenerator():
        for inName in inFileNames:
            with open(inName, 'r') as f:
                for item in json.load(f):
                    yield item

    with open(outName, 'w') as f:
        json.dump(listGenerator(), f)

我明白了：

TypeError: <generator object listGenerator at 0x7f94dc2eb3c0> is not JSON serializable

任何将所有文件加载到 ram 中的尝试都会触发 Linux 的 OOM-killer。你有什么想法吗？

【问题讨论】：

在文本之间插入逗号如何连接文档？
需要去掉每个文件的外层数组。删除每个文件的第一个字符和最后一个字符应该可以，但我想控制（并删除）json缩进。
文件实际有多大？会不会是保存完整的序列化数据比你的内存还大？
是的，这就是为什么调用 list(..) 不起作用的原因。
您是否还需要在处理 JSON 之前对其进行验证？如果不是，则无需转换字符串 -> JSON -> 字符串。只需在每个文件流之间加上逗号并用[]包围即可。

标签： python json out-of-memory generator

【解决方案1】：

您应该从list 派生并覆盖__iter__ 方法。

import json

def gen():
    yield 20
    yield 30
    yield 40

class StreamArray(list):
    def __iter__(self):
        return gen()

    # according to the comment below
    def __len__(self):
        return 1

a = [1,2,3]
b = StreamArray()

print(json.dumps([1,a,b]))

结果是[1, [1, 2, 3], [20, 30, 40]]。

【讨论】：

在 Python 2.7.8 中，StreamArray 类还必须重写 __len__ 方法并返回大于 0 的值（例如 1）。否则 json 编码器甚至不会调用 __iter__ 方法
请注意，当与 indent 参数一起使用并且可迭代对象为“空”时，此解决方案会创建无效的 JSON。 json.dumps({"products": StreamArray()}, indent=2) # {"products": ]}
如果可迭代对象为“空”，我相信我们不应该return 1 长度。

【解决方案2】：

从 simplejson 3.8.0 开始，您可以使用 iterable_as_array 选项将任何可迭代序列化为数组

# Since simplejson is backwards compatible, you should feel free to import
# it as `json`
import simplejson as json
json.dumps((i*i for i in range(10)), iterable_as_array=True)

结果是[0, 1, 4, 9, 16, 25, 36, 49, 64, 81]

【讨论】：

【解决方案3】：

根据接受的答案，这是我最终选择的 StreamArray。它包含两个谎言：

self.__tail__ 可能是不可变的建议
len(StreamArray(some_gen)) 是 0 或 1

.

class StreamArray(list):

    def __init__(self, gen):
        self.gen = gen

    def destructure(self):
        try:
            return self.__head__, self.__tail__, self.__len__
        except AttributeError:
            try:
                self.__head__ = self.gen.__next__()
                self.__tail__ = self.gen
                self.__len__ = 1 # A lie
            except StopIteration:
                self.__head__ = None
                self.__tail__ = []
                self.__len__ = 0
            return self.__head__, self.__tail__, self.__len__

    def rebuilt_gen(self):
        def rebuilt_gen_inner():
            head, tail, len_ = self.destructure()
            if len_ > 0:
                yield head
            for elem in tail:
                yield elem
        try:
            return self.__rebuilt_gen__
        except AttributeError:
            self.__rebuilt_gen__ = rebuilt_gen_inner()
            return self.__rebuilt_gen__

    def __iter__(self):
        return self.rebuilt_gen()

    def __next__(self):
        return self.rebuilt_gen()

    def __len__(self):
        return self.destructure()[2]

仅限一次性使用！

【讨论】：

+1：您的解决方案有效，但它太复杂了。我认为我实现同样的更容易。如果你发现我的劣势，看看我的。
你的看起来不错！对于我的用例，懒惰地评估第一项是一项功能。事后看来，itertools 可能会得到一些简化。很高兴知道这是按原样工作的。

【解决方案4】：

此通用解决方案对于非常大的数据也很有用”，如果结果字符串无法轻松放入内存，但仍可以轻松将其写入流来自 JSON 迭代器。（这比“import simplejson ...”更好，但不会太多）。使用 Python 2.7、3.0、3.3、3.6、3.10.0a7 测试。比simplejson 快两倍。内存占用小。书面单元测试。

import itertools

class SerializableGenerator(list):
    """Generator that is serializable by JSON"""

    def __init__(self, iterable):
        tmp_body = iter(iterable)
        try:
            self._head = iter([next(tmp_body)])
            self.append(tmp_body)
        except StopIteration:
            self._head = []

    def __iter__(self):
        return itertools.chain(self._head, *self[:1])

正常使用（输入内存很小，但仍将整个输出字符串放在内存中）：

>>> json.dumps(SerializableGenerator(iter([1, 2])))
"[1, 2]"
>>> json.dumps(SerializableGenerator(iter([])))
"[]"

对于真正的海量数据，它可以在 Python 3 中用作 JSON 块的生成器，并且仍然使用非常少的内存：

>>> iter_json = json.JSONEncoder().iterencode(SerializableGenerator(iter(range(1000000))))
>>> for chunk in iter_json:
...     stream.write(chunk)
# or a naive examle
>>> tuple(iter_json)
('[1', ', 2', ... ', 1000000', ']')

普通JSONEncoder().encode(...) 内部使用该类，json.dumps(...) 或显式JSONEncoder().iterencode(...) 使用该类来获取 JSON 块的生成器。

（示例中的函数iter() 不需要它工作，只是为了演示一个不知道长度的非平凡输入。）

测试：

import unittest
import json
# from ?your_module? import SerializableGenerator 


class Test(unittest.TestCase):

    def combined_dump_assert(self, iterable, expect):
        self.assertEqual(json.dumps(SerializableGenerator(iter(iterable))), expect)

    def combined_iterencode_assert(self, iterable, expect):
        encoder = json.JSONEncoder().iterencode
        self.assertEqual(tuple(encoder(SerializableGenerator(iter(iterable)))), expect)

    def test_dump_data(self):
        self.combined_dump_assert(iter([1, "a"]), '[1, "a"]')

    def test_dump_empty(self):
        self.combined_dump_assert(iter([]), '[]')

    def test_iterencode_data(self):
        self.combined_iterencode_assert(iter([1, "a"]), ('[1', ', "a"', ']'))

    def test_iterencode_empty(self):
        self.combined_iterencode_assert(iter([]), ('[]',))

    def test_that_all_data_are_consumed(self):
        gen = SerializableGenerator(iter([1, 2]))
        list(gen)
        self.assertEqual(list(gen), [])

此解决方案的灵感来自三个较早的答案：Vadim Pushtaev（空迭代的一些问题）和 user1158559（不必要的复杂）和 Claude（在另一个问题中，也很复杂）。

与这些解决方案的重要区别是：

重要方法 __len__、__bool__ 和其他方法一致地继承自经过有意义初始化的 list 类。
输入的第一项立即由__init__ 评估（不是由许多其他方法延迟触发）list 类可以立即知道迭代器是否为空。非空 list 包含一个带有生成器的项目，如果迭代器为空，则列表为空。
正确实现空迭代器的长度对于JSONEncoder.iterencode(...) 方法很重要。
所有其他方法都会给出有意义的输出，例如__repr__:

   >>> SerializableGenerator((x for x in range(3)))
   [<generator object <genexpr> at 0x........>]

此解决方案的一个优点是可以使用标准 JSON 序列化程序。如果应该支持嵌套生成器，那么solution with simplejson 可能是最好的，它也有与iterencode(...) 类似的变体。

存根 *.pyi 用于强类型：

from typing import Any, Iterable, Iterator

class SerializableGenerator(list):
    def __init__(self, iterable: Iterable[Any]) -> None: ...
    def __iter__(self) -> Iterator: ...

【讨论】：

【解决方案5】：

我在使用 mrjob 的 map-reduce 任务中遇到了这个错误。正确处理迭代器后得到解决。

如果您没有通过映射器处理迭代器产量，您将收到此错误。

【讨论】：