在Apache Beam中如何将嵌套重复字段合并为数组？

嘿，针对你遇到的这个Beam嵌套结构合并问题，我来给你梳理下可行的方案~

首先先从你给出的两层嵌套例子入手，再扩展到深层嵌套的场景。

两层嵌套的实现（对应你的示例） #

你想要把A1,B1,C1这类记录合并成A1: { B1: [C1, C1'], B2: C2 }的结构，其实可以通过两次分层GroupByKey + 自定义合并逻辑来实现，具体步骤如下：

1. 先把每条记录解析成((A, B), C)的键值对，这样可以先按(A,B)组合所有对应的C值；
2. 再把键拆分成A和B，转成(A, (B, C列表))的形式，然后按A分组，最后把同一A下的所有B和对应的C列表组装成嵌套字典。

用Python Beam写个示例代码更直观：

import apache_beam as beam
from typing import Tuple, Dict, List

# 模拟输入数据
input_records = [
    "A1, B1, C1",
    "A1, B1, C1'",
    "A1, B2, C2"
]

# 解析每条记录为((A,B), C)格式
class ParseToNestedKey(beam.DoFn):
    def process(self, element: str):
        a, b, c = [part.strip() for part in element.split(',')]
        yield ((a, b), c)

# 将分组后的结果转换为(A, (B, C列表))
class RepackForTopLevelGroup(beam.DoFn):
    def process(self, element: Tuple[Tuple[str, str], List[str]]):
        (a, b), c_values = element
        yield (a, (b, c_values))

# 自定义CombineFn来组装最终的嵌套字典，处理单个值/列表的情况
class BuildNestedDictCombine(beam.CombineFn):
    def create_accumulator(self):
        return {}
    
    def add_input(self, accumulator: Dict, input: Tuple[str, List[str]]):
        b_key, c_list = input
        # 如果C只有一个值就存单个元素，否则存列表
        accumulator[b_key] = c_list[0] if len(c_list) == 1 else c_list
        return accumulator
    
    def merge_accumulators(self, accumulators):
        merged_dict = {}
        for acc in accumulators:
            merged_dict.update(acc)
        return merged_dict
    
    def extract_output(self, accumulator):
        return accumulator

with beam.Pipeline() as p:
    final_result = (
        p
        | "加载输入" >> beam.Create(input_records)
        | "解析为嵌套键值对" >> beam.ParDo(ParseToNestedKey())
        | "按(A,B)分组" >> beam.GroupByKey()
        | "重组为顶层分组格式" >> beam.ParDo(RepackForTopLevelGroup())
        | "按A分组并组装嵌套结构" >> beam.CombinePerKey(BuildNestedDictCombine())
        | "打印结果" >> beam.Map(print)
    )

运行这段代码后，会输出你想要的结构：{'A1': {'B1': ['C1', "C1'"], 'B2': 'C2'}}

深层嵌套的通用解决方案 #

Beam本身并没有内置的“一键处理任意深层嵌套合并”的Transform，毕竟嵌套结构的规则太灵活了，没法做通用化封装。但我们可以沿用分层处理+逐层合并的思路来解决：

核心逻辑是：从最底层的嵌套层级开始，逐层向上做GroupByKey，每一层都把下一层的合并结果作为值，再组装到当前层级的结构里。

举个三层嵌套的例子：假设输入是A1,B1,C1,D1、A1,B1,C1,D2、A1,B1,C2,D3，想要得到A1: {B1: {C1: [D1,D2], C2: D3}}，步骤如下：

1. 先解析成((A,B,C), D)，GroupByKey得到((A,B,C), [D1,D2])；
2. 转成((A,B), (C, [D列表]))，用CombineFn合并成((A,B), {C1: [D1,D2], C2: D3})；
3. 再转成(A, (B, {C: D结构}))，最后Combine得到顶层的嵌套字典。

如果你的嵌套层级是动态的（比如不确定有多少层），还可以写一个通用工具函数：

• 先把每条记录拆分成「键层级列表」和「最终值」（比如([A,B,C], D)）；
• 从后往前逐层分组合并：每次把键列表去掉最后一位，值替换成「最后一个键+下一层的合并结果」，直到处理到最顶层的键。

总结 #

虽然Beam没有直接的深层嵌套合并Transform，但通过分层GroupByKey + 自定义CombineFn/ParDo的方式，完全可以实现任意复杂的嵌套结构合并。核心就是把复杂的嵌套拆解成一层一层的键值对分组，每一层负责组装下一层的结果，最终就能得到你想要的嵌套格式。

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