数据流如何组合 PCollections 替换对象

Posted 2023-03-31

【中文标题】数据流如何组合 PCollections 替换对象

【英文标题】：Dataflow how to combine PCollections replacing objects

【发布时间】：2018-11-24 01:05:47

【问题描述】：

我想实现一个加载 2 种数据的过程，比如说 A 类和 B 类，PCollection<A> a1, PCollection<B> b1。然后我从a1 创建一个 View.asMap() 并将其作为应用于b1 的侧输入提供给 DoFn dfn1。这个 DoFn 使用了 Kind A 的一些值并输出它们。之后，我想创建一个新的PCollection<A> a2，它包含a1 的所有对象，但替换dfn1 输出的对象。

假设a1 持有对象o1, b1, c1, d1, e1, f1, g1 dfn1 操作并输出b1 -> b2, c1 -> c2, g1 -> g2 到PCollection<A> a2

由a1 和a2 组合而成的新PCollection 应包含o1, b2, c2, e1, f1, g2

是否有内置机制来完成类似的事情？集合可以在“合并”之前键入。

提前致谢。

由于我对问题的英文解释不满意，所以这里有一个 DoFn，它执行我所要求的。真正的问题是，如果有一个内置的转换可以做这样的事情，最好不要手动创建视图。

public class CombineKvCollectionsWithMasterCollection extends DoFn<KV<String, Object>, Object>
    private static final long serialVersionUID = 4100849850259729106L;

    private PCollectionView<Map<String, Object>> masterView;

    public CombineKvCollectionsWithMasterCollection(PCollectionView<Map<String, Object>> masterView) 
        this.masterView = masterView;
    

    @ProcessElement
    public void processElement(ProcessContext c) 
        KV<String, Object> kv = c.element();
        Map<String, Object> masterMap = c.sideInput(masterView);
        if (masterMap.containsKey(kv.getKey())) 
            c.output(masterMap.get(kv.getKey()));
         else 
            c.output(kv.getValue());
        
    

【参考方案1】：

Combine 函数执行 Sum、Min、Max 和 Mean 等基本函数。对于specific combine 功能，您需要提供一些处理逻辑。所以，目前还没有内置函数可以做到这一点。