in this specific case TypeInformation tags in Flink that you seem to require even for all intermediate results.