How can I set the parallelism higher than the task slot number in more machines?

[Felipe Gutierrez]

Hi,

I deployed Flink 1.10 standalone in a cluster with 4 machines 8 cores
each. Then I configured each machine to have 8 Task Slots and
parallelism default of 8.
taskmanager.numberOfTaskSlots: 8
parallelism.default: 8
I want to run my stream app with a parallelism of 16 for each subtask.
But not having more than 8 subtasks in one TaskManager because Flink
will make them share memory [1]. I suppose that I can deploy half of
the subtasks in one machine and the second half in another machine. Is
it correct?

Then I deployed a program using "flink run -p 16 myApp.jar" and I was
monitoring it with Prometheus + Grafana and Flink created 16 subtasks
in two nodes. I mean that I am seeing 32 subtasks of each operator
that I am using. I think I was supposed to see only 16 of each
operator. Is there something wrong with my configuration?
Additional to that I also tried to use .setParallelism(16) but I got
the same result. 32 subtasks of the same operator.

Thanks,
Felipe

[1] https://ci.apache.org/projects/flink/flink-docs-release-1.10/ops/config.html#taskmanager-numberoftaskslots

--
-- Felipe Gutierrez
-- skype: felipe.o.gutierrez
-- https://felipeogutierrez.blogspot.com

[Felipe Gutierrez]

Solved! that was because I was using slotSharingGroup() in all
operators to ensure that they stay in the same task slot. I guess
Flink was creating dummy operators to ensure that.

Thanks anyway.
--
-- Felipe Gutierrez
-- skype: felipe.o.gutierrez
-- https://felipeogutierrez.blogspot.com

On Mon, May 25, 2020 at 5:54 PM Felipe Gutierrez
<[hidden email]> wrote:

>
> Hi,
>
> I deployed Flink 1.10 standalone in a cluster with 4 machines 8 cores
> each. Then I configured each machine to have 8 Task Slots and
> parallelism default of 8.
> taskmanager.numberOfTaskSlots: 8
> parallelism.default: 8
> I want to run my stream app with a parallelism of 16 for each subtask.
> But not having more than 8 subtasks in one TaskManager because Flink
> will make them share memory [1]. I suppose that I can deploy half of
> the subtasks in one machine and the second half in another machine. Is
> it correct?
>
> Then I deployed a program using "flink run -p 16 myApp.jar" and I was
> monitoring it with Prometheus + Grafana and Flink created 16 subtasks
> in two nodes. I mean that I am seeing 32 subtasks of each operator
> that I am using. I think I was supposed to see only 16 of each
> operator. Is there something wrong with my configuration?
> Additional to that I also tried to use .setParallelism(16) but I got
> the same result. 32 subtasks of the same operator.
>
> Thanks,
> Felipe
>
> [1] https://ci.apache.org/projects/flink/flink-docs-release-1.10/ops/config.html#taskmanager-numberoftaskslots
>
> --
> -- Felipe Gutierrez
> -- skype: felipe.o.gutierrez
> -- https://felipeogutierrez.blogspot.com

[Till Rohrmann]

Hi Felipe,

Flink does not create dummy operators. Unless you have configured one operator to have a parallelism of 32, you should actually only see 16 subtasks of a given operator (given that you start your program with -p 16). Be aware, though, that if you have multiple operators which cannot share the same slot, then Flink will need p_1 + p_2 + ... + p_n slots where p_i is the max parallelism of every slot sharing group you have defined in your job.

Cheers,

Till

On Mon, May 25, 2020 at 7:41 PM Felipe Gutierrez <[hidden email]> wrote:

Solved! that was because I was using slotSharingGroup() in all
operators to ensure that they stay in the same task slot. I guess
Flink was creating dummy operators to ensure that.

Thanks anyway.
--
-- Felipe Gutierrez
-- skype: felipe.o.gutierrez
-- https://felipeogutierrez.blogspot.com

On Mon, May 25, 2020 at 5:54 PM Felipe Gutierrez
<[hidden email]> wrote:
>
> Hi,
>
> I deployed Flink 1.10 standalone in a cluster with 4 machines 8 cores
> each. Then I configured each machine to have 8 Task Slots and
> parallelism default of 8.
> taskmanager.numberOfTaskSlots: 8
> parallelism.default: 8
> I want to run my stream app with a parallelism of 16 for each subtask.
> But not having more than 8 subtasks in one TaskManager because Flink
> will make them share memory [1]. I suppose that I can deploy half of
> the subtasks in one machine and the second half in another machine. Is
> it correct?
>
> Then I deployed a program using "flink run -p 16 myApp.jar" and I was
> monitoring it with Prometheus + Grafana and Flink created 16 subtasks
> in two nodes. I mean that I am seeing 32 subtasks of each operator
> that I am using. I think I was supposed to see only 16 of each
> operator. Is there something wrong with my configuration?
> Additional to that I also tried to use .setParallelism(16) but I got
> the same result. 32 subtasks of the same operator.
>
> Thanks,
> Felipe
>
> [1] https://ci.apache.org/projects/flink/flink-docs-release-1.10/ops/config.html#taskmanager-numberoftaskslots
>
> --
> -- Felipe Gutierrez
> -- skype: felipe.o.gutierrez
> -- https://felipeogutierrez.blogspot.com