複数のフローファイルを同期させる障壁のように機能するApache Nifiプロセッサ

Question

私たちのETLプロセスについてNifiを評価しています。 次のフローを作成します。 SQLデータベースから多くのデータをフェッチします - >チャンクに分割1000レコード each - >各チャンクのエラーレコードをカウント - >合計エラー数をカウントします レコードを超えた場合threshold fail process - > else各チャンクをデータベースに保存します。

解決できない問題は、すべてのチャンクが検証されるまで待つ方法です。 たとえば、5つの検証タスクを同時に実行している場合、すべてのチャンクが処理されるまで待機し、そのあと エラーカウントプロセッサを実行するまで待機する種類のバリアがあり、無効なデータを保存したくない場合は 閾値に達する。

この 検証プロセッサを複数のノードで並列に実行する可能性があり、それらがすべて完了するまで待機する可能性がある場合は、他に質問があります。

Answer 1

プロセッサーを「リリーフバルブ」として使用して、特定の属性値を持つフローファイルの最初の受信からトリガーされたメモリー内の単純なカウントを保持することができます（local/cluster stateに格納します。キーattribute-valueの値はcountになります）。その値がしきい値に達すると、完了した属性値を含むsuccess関係にルーティングするために、新しいフローファイルを生成することができます。この場合、他の結果（バッチ処理が必要なフローファイル）をMergeContentプロセッサに送信し、最小バッチングサイズを好きなものに設定します。バルブへの後続プロセッサはスケジューリング方針をEvent Drivenに設定する必要があります。したがって、バルブからフローファイルを受け取ったときにのみ実行されます。

Answer 2

分散型MapCacheの更新カウントは、フェッチと更新が別々であり、カウントを増やすアトミックプロセッサでは行えないため、正しい方法ではありません。

http://apache-nifi-users-list.2361937.n4.nabble.com/How-do-I-atomically-increment-a-variable-in-NiFi-td1084.html