Python：创建列表生成器JSON serializab

Question

如何将一个JSON文件列表转换成一个巨大的JSON数组？我有5000份文件和55万份清单。

我的第一个尝试是使用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杀手。你有什么想法吗？

Answer 1

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

# 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]

Answer 2

您应该从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]]。

Answer 3

一个完整的简单易读的解决方案，它可以将生成器从普通的或空的iterable序列化，可以与.encode（）或.iterencode（）一起使用。笔试。使用Python2.7、3.0、3.3、3.6进行测试

import itertools

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

    It is useful for serializing huge data by JSON
    >>> json.dumps(SerializableGenerator(iter([1, 2])))
    "[1, 2]"
    >>> json.dumps(SerializableGenerator(iter([])))
    "[]"

    It can be used in a generator of json chunks used e.g. for a stream
    >>> iter_json = ison.JSONEncoder().iterencode(SerializableGenerator(iter([])))
    >>> tuple(iter_json)
    ('[1', ']')
    # >>> for chunk in iter_json:
    # ...     stream.write(chunk)
    # >>> SerializableGenerator((x for x in range(3)))
    # [<generator object <genexpr> at 0x7f858b5180f8>]
    """

    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])


# -- test --

import unittest
import json


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（在另一个问题中，也很复杂）。

有用的简化是：

• 不必惰性地计算第一个项，它可以在__init__中完成，因为我们可以预期SerializableGenerator可以在json.dumps之前立即调用。（针对用户1158559解决方案）
• 不必通过NotImplementedError重写许多方法，因为这些方法并不都是像__repr__这样的方法。最好将生成器也存储到列表中，以提供有意义的结果，如[<generator object ...>]。（反对克劳德）。默认方法__len__和__bool__现在可以正确地识别空对象和非空对象。

---

此解决方案的一个优点是，可以使用标准的JSON序列化程序而无需参数。如果应该支持嵌套生成器，或者不希望使用SerializableGenerator(iterator)进行封装，那么我建议使用IterEncoder答案。