I have a flink job(2 task-managers) with a job parallelism of 64 and task slot of 64.
I have a parallelism set for one of the operators as 16. This operator(16 parallelism) slots are not getting evenly distributed across two task managers. It often takes higher task slots like 10/11 in one task manager and 5/6 in other task manager.
I'am using flink version 1.11.2. I tried adding cluster.evenly-spread-out-slots: true but it didn't work. Any solution is greatly appreciated
I don't believe there's anything more you can do. The Flink scheduler will try to co-locate consumer tasks with their producers, in order to leverage optimizations in the network stack. (Shared memory is used instead of netty whenever two communicating tasks are in the same JVM.) As I understand it, cluster.evenly-spread-out-slots: true is really only relevant for sources, which don't have input preferences.
It's often the case that reducing the parallelism for part of the job is counter-productive, as it forces a network shuffle (perhaps preventing operator chaining), and creates skew in the load on the task slots.