Apache Flink：バッファタイムアウトパラメータを変更するには？

Question

Apache Flinkはタスクの送信をバッファし、処理のために次のタスクを送信します。バッファリングはレイテンシに影響し、バッファが満たされなくても次のタスクにデータを送信するためのバッファリングのタイムアウトがあることはわかっています。

• どのようにバッファリングタイムアウトを変更できますか？私はドキュメンテーションで何も見つけることができませんでした。
• Flinkクラスタ単位またはTaskManager単位の構成ですか？タスク/オペレータごとに設定できますか？
• 同じタスクマネージャ上にタスクがあってもFlinkバッファを知っています。この状態では、同じTaskManagerにあるタスク/演算子に影響します。私は正しい？

注：このデータ/情報/質問の参照は、O'Reillyの「Streaming Processing with Apache Flink」と公式ドキュメントです。

Answer 1

バッファタイムアウトはジョブレベルパラメータで、StreamExecutionEnvironmentで設定します。

val env: StreamExecutionEnvironment = ??? 
env.setBufferTimeout(50L) // configure 50 ms buffer timeout 

オペレータごとに設定することはできません。また、すべてのオペレータの発信チャネルは、TaskManagersの送信側と受信側のペアごとに1つの物理接続に多重化されています。したがって、すべてのオペレータは、同じTaskManagerに出荷する必要がある同じバッファに出力を書き込みます。

パラメータは、documentationで詳細に説明されています。