Apache Beam单字典生成多输出问题：空PCollection求助

核心问题 #

你的代码存在三个关键问题导致空PCollection：

1. 返回值错误：_expand函数声明返回List，但实际初始化了空字典lines，且循环中未将生成的line添加到集合中，最终返回空字典，导致输出无有效数据
2. 转换类型错误：使用beam.Map（一对一转换）无法实现“单个元素拆分为多个元素”的需求，需用beam.FlatMap（一对多转换）
3. 字段处理不符合预期：未将带后缀的字段（如Market_0_30）重命名为Market、DealerMake，也未添加Age标识字段

修复后的代码 #

from typing import Dict, List
import apache_beam as beam

def _expand(element: Dict) -> List[Dict]:
    # 提取公共字段（col1、Col2、Col3等非分段字段）
    common_columns = {k: v for k, v in element.items() if not k.startswith(('Market_', 'DealerMake_'))}
    
    # 定义年龄段与对应字段的映射
    age_field_mappings = [
        ("0_30", "Market_0_30", "DealerMake_0_30"),
        ("31_60", "Market_31_60", "DealerMake_31_60"),
        ("61_90", "Market_61_90", "DealerMake_61_90"),
        ("91_120", "Market_91_120", "DealerMake_91_120"),
        ("121", "Market_121", "DealerMake_121")
    ]
    
    expanded_lines = []
    for age, market_key, dealermake_key in age_field_mappings:
        # 复制公共字段，避免修改原字典
        line = common_columns.copy()
        # 添加重命名后的字段和Age标识
        line["Market"] = element[market_key]
        line["DealerMake"] = element[dealermake_key]
        line["Age"] = age
        expanded_lines.append(line)
    
    return expanded_lines

# 使用FlatMap替代Map实现一对多转换
output = sources_data["group_stocks_view"] | "EXPAND" >> beam.FlatMap(_expand) | "PRINT" >> beam.Map(print)

关键修复点说明 #

• 提取公共字段：用字典推导式简化公共字段的筛选，避免循环判断
• 字段映射定义：明确年龄段与对应字段的关联，让逻辑更清晰
• 正确收集结果：用列表expanded_lines收集每个生成的字典，确保返回非空的结果列表
• 使用FlatMap：beam.FlatMap会将返回的列表中的每个元素单独输出为PCollection的元素，实现“一个输入拆分为5个输出”的需求
• 字段重命名与Age添加：将原分段字段重命名为统一的Market、DealerMake，并添加Age字段标识所属年龄段

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