Google Dataflowで長生きの状態

Question

ここでは、プログラミングモデルの周りに頭を抱えようとしています。シナリオ私はPub/Sub + Dataflowを使用してWebフォーラムのアナリティクスを計測しています。私は仕事がしたい

ID | TS | num_comments 
1 | 1 | 0 
1 | 2 | 1 
2 | 2 | 0 
1 | 4 | 2 

：

ID | TS | EventType 
1 | 1 | Create 
1 | 2 | Comment 
2 | 2 | Create 
1 | 4 | Comment 

そして、私がどのように見えるデータフローからの流れで終わるしたい：私は次のようになりパブ/サブから来るデータのストリームを持っていますこのロールアップはストリームプロセスとして実行され、新しいイベントが入ったときに新しいカウントが入力されます。私の質問は、ジョブが現在のトピックIDとコメントカウントの状態を保存するための慣用的な場所はどこですか？トピックは長年にわたって生きることができると仮定します。現在のアイデアは以下の通りです。

• は、私がこれを書いたとしても私はありませんBigTableのにとトピックIDの現在のコメント数がで来ているものをDoFnクエリ内のトピックIDは「現在」エントリを書きますファン。
• 何らかの方法でサイド入力を使用しますか？これは答えかもしれないようですが、もしそうなら私は完全に理解していません。
• ストリーミングジョブをグローバルウィンドウで設定します。トリガーはレコードを取得するたびにオフになり、データフローを使用してペインの履歴全体をどこかに保持します。 （無制限のストレージ要件ですか？）

EDIT：これらの3つの戦略、またはそれを実行する他のさまざまな方法のいずれかを実装するのに問題はありません。 最高の Dataflowでこれを行う方法。バックフィルの履歴を再処理する必要があるもの

EDIT2：データフローサービスにバグがあり、現在、フラットトランスフォームに入力を追加すると更新が失敗するというバグがありますフラット操作に何かを追加することを含むジョブを変更した場合、そのジョブで発生した状態を破棄して再構築する必要があります。

Answer 1

これを実行するには、トリガーと結合を使用できるようにする必要があります。

PCollection<ID> comments = /* IDs from the source */; 
PCollection<KV<ID, Long>> commentCounts = comments 
    // Produce speculative results by triggering as data comes in. 
    // Note that this won't trigger after *every* element, but it will 
    // trigger relatively quickly (as the system divides incoming data 
    // into work units). You could also throttle this with something 
    // like: 
    // AfterProcessingTime.pastFirstElementInPane() 
    //  .plusDelayOf(Duration.standardMinutes(5)) 
    // which will produce output every 5 minutes 
    .apply(Window.triggering(
      Repeatedly.forever(AfterPane.elementCountAtLeast(1))) 
     .accumulatingFiredPanes()) 
    // Count the occurrences of each ID 
    .apply(Count.perElement()); 

// Produce an output String -- in your use case you'd want to produce 
// a row and write it to the appropriate source 
commentCounts.apply(new DoFn<KV<ID, Long>, String>() { 
    public void processElement(ProcessContext c) { 
    KV<ID, Long> element = c.element(); 
    // This includes details about the pane of the window being 
    // processed, and including a strictly increasing index of the 
    // number of panes that have been produced for the key.   
    PaneInfo pane = c.pane(); 
    return element.key() + " | " + pane.getIndex() + " | " + element.value(); 
    } 
}); 

あなたのデータに応じて、また、ソースからの全体のコメントを読んIDを抽出し、各IDのカウントを取得するためにCount.perKey()を使用することができます。より複雑な組み合わせが必要な場合は、カスタムCombineFnを定義し、Combine.perKeyを使用して見ることができます。

Answer 2

のBigQueryが上書き行をサポートしていないので、これについて移動する一つの方法は、BigQueryのにイベントを書き込み、COUNTを使用してデータを照会することです：

SELECT ID、ID BY表GROUPからCOUNT（NUM_COMMENTS）。

BigQueryにエントリを書き込む前に、Dataflow内のnum_commentsのウィンドウごとの集計を行うこともできます。上記のクエリは引き続き機能します。