如何在Python的Apache Beam中读取字典列表并拆分单个字典？

要实现读取存储字典列表的文本文件，并逐个返回列表中的字典，可以按以下步骤操作：

核心思路 #

1. 读取文本文件，获取完整的JSON列表字符串
2. 将JSON字符串解析为Python字典列表
3. 展开列表，将每个字典作为独立元素输出

代码实现 #

import apache_beam as beam
import json

def process_dict_list_file(input_file, output_path):
    with beam.Pipeline() as pipeline:
        (
            pipeline
            # 读取目标文本文件，每行作为一个元素（此处文件为单行JSON列表）
            | "读取文本文件" >> beam.io.ReadFromText(input_file)
            # 解析JSON字符串为Python字典列表
            | "解析JSON列表" >> beam.Map(json.loads)
            # 展开列表，将每个字典单独输出
            | "拆分单个字典" >> beam.FlatMap(lambda dict_list: dict_list)
            # 可选：将结果写入文本文件，或替换为其他处理逻辑（如打印、写入数据库等）
            | "输出结果" >> beam.io.WriteToText(output_path, shard_name_template='')
        )

if __name__ == "__main__":
    # 替换为你的输入文件路径和输出路径
    process_dict_list_file("input.json", "output.txt")

代码说明 #

• beam.io.ReadFromText：按行读取文件内容，若你的文件中每行都是一个独立的字典列表，该操作也能自动处理每行数据
• beam.Map(json.loads)：将每行的JSON字符串转换为Python的字典列表对象
• beam.FlatMap：将列表展开，把原本的单个列表元素拆分为多个字典元素，后续步骤即可对每个字典单独进行处理

内容的提问来源于stack exchange，提问作者Amarjeet Kushwaha