如何使用 Apache Beam 将列表对象写入 JSON 文件？答案

【问题标题】：How to write list object into a JSON file using apache beam?如何使用 Apache Beam 将列表对象写入 JSON 文件？
【发布时间】：2021-10-10 09:31:40
【问题描述】：

我有一个字典元素列表，如下所示。

list_data = [
    {"id":"1", "name":"Cow", "type": "animal"},
    {"id":"2", "name":"Lion", "type": "animal"},
    {"id":"3", "name":"Peacock", "type": "bird"},
    {"id":"4", "name":"Giraffe", "type": "animal"}
]

我希望使用 apache 光束管道将上述列表写入 JSON 文件。

我试过这样做：

class BeamProcess:

    def process_data():

        json_file_path = "gs://my_bucket/df_output/output.json"
        
        list_data = [
            {"id":"1", "name":"Cow", "type": "animal"},
            {"id":"2", "name":"Lion", "type": "animal"},
            {"id":"3", "name":"Peacock", "type": "bird"},
            {"id":"4", "name":"Giraffe", "type": "animal"}
        ]

        argv = [
                '--project=<my_project>',
                '--region=<region>',
                '--job_name=<custom_name>',
                '--temp_location=<temporary_location>',
                '--runner=DataflowRunner'
            ]

        p = beam.Pipeline(argv=argv)

        (
                p
                | 'Create' >> beam.Create(list_data)
                | 'Write Output' >> beam.io.WriteToText(json_file_path, shard_name_template='')
        )
        p.run().wait_until_finish()


if __name__ == "__main__":
    beam_proc = BeamProcess()
    beam_proc.process_data()

当我执行上述代码时，我最终在 output.json 文件中看到以下行。

{"id":"1", "name":"Cow", "type": "animal"}
{"id":"2", "name":"Lion", "type": "animal"}
{"id":"3", "name":"Peacock", "type": "bird"}
{"id":"4", "name":"Giraffe", "type": "animal"}

但我希望看到的是：

[
    {"id":"1", "name":"Cow", "type": "animal"},
    {"id":"2", "name":"Lion", "type": "animal"},
    {"id":"3", "name":"Peacock", "type": "bird"},
    {"id":"4", "name":"Giraffe", "type": "animal"}
]

使用 apache beam 将 python 列表对象写入 JSON 文件的正确方法是什么？

【问题讨论】：

我怀疑 Create 转换将您的列表解释为多个元素，而不是将列表用作单个元素。因此，当您编写输出时，您只需按顺序写出每个元素。如果在将列表传递给 Create 之前将列表嵌套在第二个列表中，您能看到会发生什么吗？
嗨@DanielOliveira，将列表嵌套在第二个列表中是有效的，我现在可以在 output.json 文件中看到预期的列表。谢谢。
没问题。既然它有效，我也会把它作为答案。

标签： python-3.x google-cloud-platform google-cloud-dataflow apache-beam

【解决方案1】：

当给beam.Create 一个列表时，它会将其解释为生成的PCollection 的元素列表。当您将 PCollection 写成文本时，您输出的是四个单独的元素而不是一个列表，这就是它未按预期格式化的原因。

beam.Create([1, 2, 3, 4]) # Creates a PCollection of four int elements.

因此，为了创建一个包含列表作为元素的 PCollection，您需要嵌套要用作元素的列表，如下所示：

beam.Create([[1, 2, 3, 4]]) # Creates a PCollection of one list element.

【讨论】：