Copy/pasting a lot in Apache Flink jobs

This is more a code refactoring question and Im seeking suggestion on how best to improve code reusability.