GoogleのデータフローでパーソナライズされたWindowFnを作成する方法

Question

WindowFnを別の方法で作成して、入力項目のタイムスタンプに基づいてではなく、別の項目に基づいて入力要素にWindowsを割り当てたいとします。 Google DataFlow SDKのあらかじめ定義されたWindowFnは、ウィンドウを割り当てる基準としてタイムスタンプを使用しています。

具体的には、SlidingWindowsの種類を作成したいと思いますが、ウィンドウ割り当ての基準としてタイムスタンプを考慮する代わりに、その基準として別のフィールドを検討したいと思います。

カスタマイズしたWindowFnを作成するにはどうすればよいですか？自分自身でWindowFnを作成する際に考慮する必要がある点は何ですか？

ありがとうございました。

Answer 1

新しいWindowFnを作成するには、WindowFnまたはサブクラスから継承し、さまざまな抽象メソッドをオーバーライドするだけです。

あなたのケースでは、あなたがNonMergingWindowFnから継承することができるようにあなたは、ウィンドウのマージを必要としない、とあなたのコードが

public class MyWindowFn extends NonMergingWindowFn<ElementT, IntervalWindow> { 
    public Collection<W> assignWindows(AssignContext c) { 
    return setOfWindowsElementShouldBeIn(c.element()); 
    } 

    public boolean isCompatible(WindowFn other) { 
    return other instanceof MyWindowFn; 
    } 

    public Coder<IntervalWindow> windowCoder() { 
    return IntervalWindow.getCoder(); 
    } 

    public W getSideInputWindow(final BoundedWindow window) { 
    // You may not need this if you won't ever be using PCollections windowed 
    // with this as side inputs. If that's the case, just throw. 
    // Otherwise you'll need to figure out how to map the main input windows 
    // into the windows generated by this WindowFn. 
    } 
} 

のようになります