如何使用Apache Beam将Python列表对象写入标准JSON文件

你当前输出的是JSON Lines格式（每行一个独立JSON对象），这是WriteToText的默认行为：每个输入元素会被单独转为字符串后逐行写入，不会自动添加数组包裹符和元素间的分隔逗号。要输出标准JSON数组，需要先将所有元素聚合为完整列表，再整体序列化为JSON格式后写入。

修改后的完整代码 #

import apache_beam as beam
import json

class BeamProcess:
    # 补充self参数，否则实例调用会报错
    def process_data(self):
        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)
                # 第一步：将所有元素聚合为一个全局列表
                | 'Aggregate to List' >> beam.combiners.ToList()
                # 第二步：将整个列表序列化为带缩进的标准JSON字符串
                | 'Serialize to JSON' >> beam.Map(lambda x: json.dumps(x, indent=4, ensure_ascii=False))
                | '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()

注意事项 #

• 该方案会将全量数据汇聚到单个Worker节点处理，仅适用于小数据集场景。如果数据规模较大，更推荐使用你当前得到的JSON Lines格式：该格式天然支持分布式读写，不需要加载全量文件到内存即可逐行解析，是大数据场景下的通用标准。
• 若数据量极大又必须输出标准JSON数组，建议先写入JSON Lines格式，再单独运行后处理任务将其转换为数组格式，避免Beam流水线出现内存溢出问题。

内容的提问来源于stack exchange，提问作者Gopinath S