如何在Apache Beam（Java）中并行合并同标识JSON数据

原始输入为两条JSON记录，它们的companyId和asOfDate完全一致，仅data数组内容不同：

{"companyId":"xxxx", "asOfDate":"2022-12-11T00:00:00", "data":[{"demographicVariable":"Ethnicity", "value": "23" }]}
{"companyId":"xxxx", "asOfDate":"2022-12-11T00:00:00", "data":[{"demographicVariable":"Age Band", "value": 10}]}

需要合并为单条JSON，将两个data数组合并到一起：

{"companyId":"xxxx", "asOfDate":"2022-12-11T00:00:00", "data":[{"demographicVariable":"Age Band", "value": 10},{"demographicVariable":"Ethnicity", "value": "23" }]}

Beam的并行处理核心是按键分组聚合，我们可以利用这个特性高效合并同维度数据，具体实现步骤如下：

1. 定义数据模型POJO #

先定义对应JSON结构的Java类，方便序列化/反序列化和数据操作：

import com.fasterxml.jackson.annotation.JsonIgnoreProperties;
import java.util.List;

@JsonIgnoreProperties(ignoreUnknown = true)
public class DemographicRecord {
    private String companyId;
    private String asOfDate;
    private List<DemographicData> data;

    // 无参构造器、Getter、Setter方法
    public static class DemographicData {
        private String demographicVariable;
        private Object value;

        // 无参构造器、Getter、Setter方法
    }
}

2. 构建Beam Pipeline #

核心流程 #

• 读取输入JSON文本并解析为POJO对象
• 以companyId + asOfDate为分组键，将同维度数据归为一组（Beam自动并行处理不同分组）
• 合并每组内的data数组
• 将合并后的对象序列化为JSON并输出

完整代码示例 #

import org.apache.beam.sdk.Pipeline;
import org.apache.beam.sdk.io.TextIO;
import org.apache.beam.sdk.transforms.*;
import org.apache.beam.sdk.values.KV;
import org.apache.beam.sdk.values.PCollection;
import com.fasterxml.jackson.databind.ObjectMapper;

public class JsonMergePipeline {
    private static final ObjectMapper objectMapper = new ObjectMapper();

    public static void main(String[] args) {
        Pipeline pipeline = Pipeline.create();

        // 1. 读取输入JSON文件
        PCollection<String> inputJson = pipeline.apply(TextIO.read().from("path/to/input/json"));

        // 2. 解析JSON为POJO并生成分组键
        PCollection<KV<String, DemographicRecord>> keyedRecords = inputJson.apply(ParDo.of(new DoFn<String, KV<String, DemographicRecord>>() {
            @ProcessElement
            public void processElement(ProcessContext c) {
                try {
                    DemographicRecord record = objectMapper.readValue(c.element(), DemographicRecord.class);
                    String groupKey = record.getCompanyId() + "_" + record.getAsOfDate();
                    c.output(KV.of(groupKey, record));
                } catch (Exception e) {
                    // 可根据需求添加异常日志或跳过错误数据
                    e.printStackTrace();
                }
            }
        }));

        // 3. 按分组键聚合并合并data数组
        PCollection<DemographicRecord> mergedRecords = keyedRecords.apply(GroupByKey.create())
                .apply(ParDo.of(new DoFn<KV<String, Iterable<DemographicRecord>>, DemographicRecord>() {
                    @ProcessElement
                    public void processElement(ProcessContext c) {
                        Iterable<DemographicRecord> groupRecords = c.element().getValue();
                        DemographicRecord merged = null;
                        for (DemographicRecord record : groupRecords) {
                            if (merged == null) {
                                merged = new DemographicRecord();
                                merged.setCompanyId(record.getCompanyId());
                                merged.setAsOfDate(record.getAsOfDate());
                                merged.setData(record.getData());
                            } else {
                                merged.getData().addAll(record.getData());
                            }
                        }
                        if (merged != null) {
                            c.output(merged);
                        }
                    }
                }));

        // 4. 序列化合并后的对象为JSON并输出
        mergedRecords.apply(ParDo.of(new DoFn<DemographicRecord, String>() {
            @ProcessElement
            public void processElement(ProcessContext c) {
                try {
                    String mergedJson = objectMapper.writeValueAsString(c.element());
                    c.output(mergedJson);
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
        })).apply(TextIO.write().to("path/to/output/merged.json").withoutSharding());

        pipeline.run().waitUntilFinish();
    }
}

并行特性说明 #

• GroupByKey是Beam并行聚合的核心，框架会自动将不同分组的数据分配到多个Worker节点并行处理，仅同分组的数据会被汇聚合并
• 解析、分组、合并、序列化等各个阶段均可并行执行，充分利用集群资源提升处理效率

内容的提问来源于stack exchange，提问作者Duany Baró Menéndez