問題タブ [spotify-scio]

pubsub からバイト配列を読み取り、それらをウィンドウ化し、GCS のテキスト ファイルに書き込む単純なデータフロー パイプラインを作成しました。トラフィックの少ないトピックではこれが完全に機能することがわかりましたが、1 分あたり約 2.4GB のトピックで実行したところ、いくつかの問題が発生し始めました。

パイプラインを開始するとき、ワーカーの数を設定していませんでした (必要に応じて自動スケーリングされると想像していたので)。この量のデータを取り込むとき、ワーカーの数は 1 のままでしたが、TextIO.write() は 2 分のウィンドウを書き込むのに 15 分以上かかっていました。これは、メモリがなくなるまでバックアップされ続けます。このステップがバックアップされたときに Dataflow が自動スケーリングしない正当な理由はありますか?

ワーカーの数を 6 に増やしたとき、ファイルの書き込み時間は 5 分間のウィンドウで約 4 分から始まり、その後わずか 20 秒まで短縮されました。

また、ワーカを6体使用する場合、ウォールタイムの計算に問題がありそうですか？データフローが追いついたとしても、私のものは決してダウンするようには見えず、4時間実行した後、書き込みステップの要約は次のようになりました。

ジョブ ID: 2019-03-13_19_22_25-14107024023503564121

scala spotify scio を使用して、ビーム プログラミング プロジェクトの入力として .txt ファイルがあります。

input= args.getOrElse("input", "/home/user/Downloads/trade-20181001.txt")

ファイル名から日付 2018-10-01 (10 月 1 日) を抽出するにはどうすればよいですか? ありがとうございました！

目的：

ストリーム データを読み込み、キーを追加して、キーごとにカウントしたい。

問題：

ストリーミング アプローチ (無制限のデータ) を使用して大きなサイズのデータ​​を読み込んでキーごとにグループ化しようとすると、Apache Beam Dataflow パイプラインでメモリ エラーが発生します。データがグループバイに蓄積されているようで、各ウィンドウのトリガーでデータを先に発射しないためです。

要素のサイズを小さくすると (要素の数は変わりません)、うまくいきます! 実際には group-by step はすべてのデータがグループ化されるのを待ってから、すべての新しいウィンドウ化されたデータを起動するためです。

私は両方でテストしました：

ビーム バージョン 2.11.0 および scio バージョン 0.7.4

ビーム バージョン 2.6.0 および scio バージョン 0.6.1

エラーを再生成する方法:

1. ファイル名を含む Pubsub メッセージを読み取る
2. 関連ファイルをGCSから行ごとの反復子として読み取ってロードします
3. 行ごとにフラット化する (約 10,000 の要素を生成する)
4. 要素にタイムスタンプ (現在の時刻) を追加する
5. データのキー値を作成します (1 から 10 までのランダムな整数キーを使用)
6. トリガーを使用してウィンドウを適用します (行が小さく、メモリに問題がない場合、約 50 回トリガーされます)
7. キーごとにカウントします（キーごとにグループ化してから結合します）
8. 最後に、ウィンドウとキーごとのカウントを表す約 50 * 10 の要素があると想定しました (行サイズが十分に小さい場合、正常にテストされました)。

パイプラインの視覚化 (ステップ 4 から 7 ):

group-by-key ステップの要約:

ご覧のとおり、データはグループごとに蓄積され、出力されません。

ウィンドウ処理コードは次のとおりです。

エラー：

グループバイに各ウィンドウの初期結果を強制的に発行させることで、メモリの問題を解決する解決策はありますか。