私は、すべてが1台のマシンに割り当てられた最新のものまで、タスク（ステップ？ジョブ？）をかなり均等に分散していた3ノードクラスターをセットアップしました。
トポロジ (まだ flink にこの用語を使用していますか?):
kafka (3 topics on different feeds) -> flatmap -> union -> map
このセットアップについて、クラスター マネージャーにすべてを 1 台のマシンに配置するように指示する何かがありますか?
また、画像の「設定されていない」値は何ですか? 私が逃したいくつかのステップ？または、実装される予定の UI 機能はありますか?
実際には、Flink が単一の TaskManager でジョブをスケジュールするのは意図的です。それを理解するために、Flink のリソース スケジューリング アルゴリズムを簡単に説明します。
まず第一に、Flink の世界では、スロットは複数のタスク (オペレーターの並列インスタンス) に対応できます。実際、各オペレーターの 1 つの並列インスタンスに対応できます。この理由は、Flink がストリーミング ジョブをストリーミング形式で実行するだけでなく、バッチ ジョブも実行するためです。ストリーミング方式とは、Flink がデータフロー グラフのすべての演算子をオンラインにして、中間結果を下流の演算子に直接ストリーミングして消費できるようにすることを意味します。デフォルトでは、Flink は各オペレーターの 1 つのタスクを 1 つのスロットに結合しようとします。
Flink がタスクを異なるスロットにスケジュールするとき、不必要なネットワーク通信を避けるためにタスクを入力と同じ場所に配置しようとします。ソースの場合、コロケーションは実装によって異なります。たとえば、ファイルベースのソースの場合、Flink はローカル ファイル入力分割をさまざまなタスクに割り当てようとします。
これをあなたの仕事に当てはめると、次のことがわかります。並列処理 1 の 3 つの異なるソースがあります。すべてのソースは同じリソース共有グループに属しているため、各オペレーターの 1 つのタスクが同じスロットにデプロイされます。最初のスロットは、利用可能なインスタンスからランダムに選択され (実際には、 でのTaskManager登録の順序によって異なりますJobManager)、埋められます。選択したスロットが machine にあるとしましょうnode1。
次に、並列度が 2 の 3 つのフラット マップ オペレーターがあります。ここでも、各フラット マップ オペレーターの 2 つのサブタスクの 1 つを、既に 3 つのソースに対応している同じスロットにデプロイできます。ただし、2 番目のサブタスクは新しいスロットに配置する必要があります。これが発生すると、Flink はタスクの入力の 1 つがデプロイされているスロットと同じ場所にある空きスロットを選択しようとします (これもネットワーク通信を減らすためです)。の 1 つのスロットのみnode1が占有されて31いるため、まだ空いているため、各 flatMap オペレーターの 2 番目のサブタスクも に展開しnode1ます。
同じことがタンブリング ウィンドウの縮小操作にも適用されます。Flink は、ウィンドウ オペレーターのすべてのタスクをその入力と同じ場所に配置しようとします。そのすべての入力が実行されnode1、node1ウィンドウ オペレーターの 6 つのサブタスクに対応するのに十分な空きスロットがあるため、それらは にスケジュールされnode1ます。1 つのウィンドウ タスクが、3 つのソースと各 flatMap オペレーターの 1 つのタスクを含むスロットで実行されることに注意することが重要です。
これで、Flink がジョブの実行に 1 台のマシンのスロットのみを使用する理由が説明されることを願っています。
問題は、キーのない (グループ化されていない) ストリームでグローバル ウィンドウを構築しているため、ウィンドウを 1 台のマシンで実行する必要があることです。
ストリームをグループ化できるように、アプリケーション ロジックを別の方法で表現することもできます。
"(not set)" の部分はおそらく Flink のDataStreamAPI の問題であり、デフォルトのオペレーター名を設定していません。APIに対して実装されたジョブは次のDataSetようになります。
