Flink分散プログラムにおける例外処理

[[409230]]

この記事はWeChatの公開アカウント「易言」から転載したもので、著者は易言です。この記事の転載についてはYiyan公式アカウントまでご連絡ください。

当社のデータ プラットフォーム製品では、開発を簡素化するために、Flink をカプセル化し、ジョブとフローの抽象化を定義しています。ジョブは実際には Flink 操作です。各ジョブは複数のフローを定義できます。 Flow は、Flink の DataStream として理解できます。ジョブによって渡される StreamExecutionEnvironment を使用すると、ソースやシンクを含む複数のオペレーターをフローに追加できます。

ジョブとフローの関係はカスタム @JobFlow アノテーションを使用して構成できるため、抽象 AbstractJob の run() メソッドを実行するときに、リフレクションを使用してジョブの下にあるすべてのフローを取得し、各フローの run() メソッドをトラバースして実行できます。 Flow の run() メソッドでは、実際には StreamExecutionEnvironment に従って複数の演算子が実行されます。

コンピューティングの安定性を確保するために、Flink はさまざまな再起動戦略を提供します。たとえば、再起動戦略を失敗率に設定すると、実行されたタスクのエラー数が設定された失敗率に達すると、Flink Worker ノードの TaskManager が再起動されます。再起動回数を超えると、タスク マネージャーは実行を停止します。

失敗の原因は、リソース不足、ネットワーク通信障害、Flink クラスター環境に起因するその他の障害など、さまざまなものが考えられます。ただし、作成したジョブが不適切なデータ処理のためにストリーミング データを処理するときにビジネス例外をスローし、Flink がそれを失敗と見なした可能性もあります。

ビジネス上の理由でスローされた例外によるタスク マネージャーの不要な再起動を減らすには、作成する Flink プログラムの例外処理メカニズムを指定する必要があります。 Flink のジョブはカプセル化されているため、最初からビジネス例外の問題を一度に解決することを検討しました。つまり、AbstractJob の run() メソッドでカスタム ビジネス例外をキャプチャし、エラー情報がログに記録された後に例外を「処理」して、実行の失敗につながる例外のスローを回避し、TaskManager の再起動を引き起こします。例:

パブリック抽象クラスAbstractFlowはFlowを実装します{
パブリックボイド実行（）{
        試す {
            実行Bare();
        } キャッチ (DomainException ex) {
            //...
        }
    } 
   
    保護された抽象 void runBare();
 }

しかし、この処理メカニズムではビジネス例外をまったく捕捉できません。なぜ？これは、Flink の分散メカニズムから始まります。

Flink クラスターでタスクを実行するには、クライアントが Flink クラスターのマスター ノードにジョブを送信する必要があります。マスターのディスパッチャはジョブを受信して​​ JobManager を起動します。JobManager はジョブの論理ビューを分析してジョブのリソース要件を理解し、このジョブに必要なリソースを ResourceManager (スタンドアロン モード。YARN の場合は、リソースは YARN によって管理およびスケジュールされます) に適用します。 JobManager はジョブの論理ビューを物理ビューに変換し、コンピューティング タスクを Flink クラスターの TaskManager に配布します。実行プロセス全体を次の図に示します。

物理的な観点から見ると、カプセル化するフローは実際にはジョブ、つまり前述のコンピューティング タスクです。ジョブには複数のオペレーターを含めることができます。隣接する演算子間でデータのシャッフルがなく、並列度が同じである場合、それらは演算子チェーンにマージされます。各オペレーターまたはオペレーター チェーンは JobVertex を形成し、実行中にタスクとして機能します。並列処理の設定に応じて、各タスクには同じ並列数のサブタスクが含まれます。これらのサブタスクは、ジョブ スケジューリングの最小の論理単位であり、プロセス リソース内のスレッドに対応し、Flink 内のスロットになります (スロット共有が考慮されていない場合)。

Flink 環境の並列度が 1 に設定され、ジョブの最初の 2 つのオペレーターがマージ オペレーター チェーンの要件を満たし、並列度が 2 に設定されていると仮定します。次に、データは keyBy() などの演算子を通じてシャッフルされ、同じ Sink にマージされます。それらの関係は以下の図に示されます。

当然ですが、Flink クラスターはジョブを実行する際にジョブを分割し、分割されたサブタスクを TaskManager 内の各スロットに分配します。 TaskManager は JVM であり、Slot はプロセス内のスレッドです。

答えは自明です。 AbstractFlow がタスク実行時に各オペレーターによってスローされたビジネス例外をキャプチャできない理由は、それらが同じ JVM または同じスレッドで実行されていないためです。これが分散開発とローカル開発の本質的な違いです。 Flink の実行原理を理解していないと、Java の例外処理メカニズムがなぜ機能しないのか混乱してしまうかもしれません。分散開発を行う際に、ローカル開発の経験をそのままコピーしていると、真実を知るまでに本当に頭をぶつけて血を流さなければならないかもしれません。したがって、正しいアプローチは、各演算子の実装で各例外をキャプチャすること、つまり、各演算子自体が堅牢であることを確認して、ジョブが可能な限り堅牢であることを保証することです。

もちろん、分散開発とローカル開発の本質的な違いはこれに限りません。たとえば、分散開発におけるプロセス間呼び出しのシリアル化の要件、データの一貫性に関するさまざまな要件、非同期通信メカニズムとブロッキング呼び出しの理解は、プログラマーにさまざまな経験をもたらす可能性があります。結局のところ、分散開発や分散システムの基本原理を理解することで、できるだけ早く真実を把握し、知らないうちに落とし穴に陥ることを避けることができます。