[DISCUSS] FLINK-21110: Optimize Scheduler Performance for Large-Scale Jobs

[Zhilong Hong]

Hello, everyone:

I would like to start the discussion about FLINK-21110: Optimize Scheduler Performance for Large-Scale Jobs [1].

According to the result of scheduler benchmarks we implemented in FLINK-20612 [2], the bottleneck of deploying and running a large-scale job in Flink is mainly focused on the following procedures:

Procedure       Time complexity
Initializing ExecutionGraph
O(N^2)
Building DefaultExecutionTopology
O(N^2)
Initializing PipelinedRegionSchedulingStrategy
O(N^2)
Scheduling downstream tasks when a task finishes
O(N^2)
Calculating tasks to restart when a failover occurs
O(N^2)
Releasing result partitions
O(N^2)

These procedures are all related to the complexity of the topology in the ExecutionGraph. Between two vertices connected with the all-to-all edges, all the upstream Intermediate ResultPartitions are connected to all downstream ExecutionVertices. The computation complexity of building and traversing all these edges will be O(N^2).

As for memory usage, currently we use ExecutionEdges to store the information of connections. For the all-to-all distribution type, there are O(N^2) ExecutionEdges. We test a simple job with only two vertices. The parallelisms of them are both 10k. Furthermore, they are connected with all-to-all edges. It takes 4.175 GiB (estimated via MXBean) to store the 100M ExecutionEdges.

In most large-scale jobs, there will be more than two vertices with large parallelisms, and they would cost a lot of time and memory to deploy the job.

As we can see, for two JobVertices connected with the all-to-all distribution type, all IntermediateResultPartitions produced by the upstream ExecutionVertices are isomorphic, which means that the downstream ExecutionVertices they connected are exactly the same. The downstream ExecutionVertices belonging to the same JobVertex are also isomorphic, as the upstream ResultPartitions they connect are the same, too.

Since every JobEdge has exactly one distribution type, we can divide the vertices and result partitions into groups according to the distribution type of the JobEdge.

For the all-to-all distribution type, since all downstream vertices are isomorphic, they belong to a single group, and all the upstream result partitions are connected to this group. Vice versa, all the upstream result partitions also belong to a single group, and all the downstream vertices are connected to this group. In the past, when we wanted to iterate all the downstream vertices, we needed to loop over them n times, which leads to the complexity of O(N^2). Now since all upstream result partitions are connected to one downstream group, we just need to loop over them once, with the complexity of O(N).

For the pointwise distribution type, because each result partition is connected to different downstream vertices, they should belong to different groups. Vice versa, all the vertices belong to different groups. Since one result partition group is connected to one vertex group pointwisely, the computation complexity of looping over them is still O(N).

After we group the result partitions and vertices, ExecutionEdge is no longer needed. For the test job we mentioned above, the optimization can effectively reduce the memory usage from 4.175 GiB to 12.076 MiB (estimated via MXBean) in our POC. The time cost is reduced from 62.090 seconds to 8.551 seconds (with 10k parallelism).

The detailed design doc with illustrations is located at [3]. Please find more details in the links below.

Looking forward to your feedback.

[1] https://issues.apache.org/jira/browse/FLINK-21110
[2] https://issues.apache.org/jira/browse/FLINK-20612
[3] https://docs.google.com/document/d/1OjGAyJ9Z6KsxcMtBHr6vbbrwP9xye7CdCtrLvf8dFYw/edit?usp=sharing

[Till Rohrmann]

Thanks for making the community aware of these performance improvements,
Zhilong. I like them and I am looking forward to a faster Flink :-)

Cheers,
Till

On Tue, Feb 2, 2021 at 11:00 AM Zhilong Hong <[hidden email]> wrote:

> Hello, everyone:
>
> I would like to start the discussion about FLINK-21110: Optimize Scheduler
> Performance for Large-Scale Jobs [1].
>
> According to the result of scheduler benchmarks we implemented in
> FLINK-20612 [2], the bottleneck of deploying and running a large-scale job
> in Flink is mainly focused on the following procedures:
>
> Procedure       Time complexity
> Initializing ExecutionGraph
> O(N^2)
> Building DefaultExecutionTopology
> O(N^2)
> Initializing PipelinedRegionSchedulingStrategy
> O(N^2)
> Scheduling downstream tasks when a task finishes
> O(N^2)
> Calculating tasks to restart when a failover occurs
> O(N^2)
> Releasing result partitions
> O(N^2)
>
> These procedures are all related to the complexity of the topology in the
> ExecutionGraph. Between two vertices connected with the all-to-all edges,
> all the upstream Intermediate ResultPartitions are connected to all
> downstream ExecutionVertices. The computation complexity of building and
> traversing all these edges will be O(N^2).
>
> As for memory usage, currently we use ExecutionEdges to store the
> information of connections. For the all-to-all distribution type, there are
> O(N^2) ExecutionEdges. We test a simple job with only two vertices. The
> parallelisms of them are both 10k. Furthermore, they are connected with
> all-to-all edges. It takes 4.175 GiB (estimated via MXBean) to store the
> 100M ExecutionEdges.
>
> In most large-scale jobs, there will be more than two vertices with large
> parallelisms, and they would cost a lot of time and memory to deploy the
> job.
>
> As we can see, for two JobVertices connected with the all-to-all
> distribution type, all IntermediateResultPartitions produced by the
> upstream ExecutionVertices are isomorphic, which means that the downstream
> ExecutionVertices they connected are exactly the same. The downstream
> ExecutionVertices belonging to the same JobVertex are also isomorphic, as
> the upstream ResultPartitions they connect are the same, too.
>
> Since every JobEdge has exactly one distribution type, we can divide the
> vertices and result partitions into groups according to the distribution
> type of the JobEdge.
>
> For the all-to-all distribution type, since all downstream vertices are
> isomorphic, they belong to a single group, and all the upstream result
> partitions are connected to this group. Vice versa, all the upstream result
> partitions also belong to a single group, and all the downstream vertices
> are connected to this group. In the past, when we wanted to iterate all the
> downstream vertices, we needed to loop over them n times, which leads to
> the complexity of O(N^2). Now since all upstream result partitions are
> connected to one downstream group, we just need to loop over them once,
> with the complexity of O(N).
>
> For the pointwise distribution type, because each result partition is
> connected to different downstream vertices, they should belong to different
> groups. Vice versa, all the vertices belong to different groups. Since one
> result partition group is connected to one vertex group pointwisely, the
> computation complexity of looping over them is still O(N).
>
> After we group the result partitions and vertices, ExecutionEdge is no
> longer needed. For the test job we mentioned above, the optimization can
> effectively reduce the memory usage from 4.175 GiB to 12.076 MiB (estimated
> via MXBean) in our POC. The time cost is reduced from 62.090 seconds to
> 8.551 seconds (with 10k parallelism).
>
> The detailed design doc with illustrations is located at [3]. Please find
> more details in the links below.
>
> Looking forward to your feedback.
>
> [1] https://issues.apache.org/jira/browse/FLINK-21110
> [2] https://issues.apache.org/jira/browse/FLINK-20612
> [3]
> https://docs.google.com/document/d/1OjGAyJ9Z6KsxcMtBHr6vbbrwP9xye7CdCtrLvf8dFYw/edit?usp=sharing
>
>

[Guowei Ma]

Thanks to ZhiLong for improving the scheduler's performance.
I think many users would benefit from your work!
Best,
Guowei


On Wed, Feb 3, 2021 at 2:04 AM Till Rohrmann <[hidden email]> wrote:

> Thanks for making the community aware of these performance improvements,
> Zhilong. I like them and I am looking forward to a faster Flink :-)
>
> Cheers,
> Till
>
> On Tue, Feb 2, 2021 at 11:00 AM Zhilong Hong <[hidden email]>
> wrote:
>
> > Hello, everyone:
> >
> > I would like to start the discussion about FLINK-21110: Optimize
> Scheduler
> > Performance for Large-Scale Jobs [1].
> >
> > According to the result of scheduler benchmarks we implemented in
> > FLINK-20612 [2], the bottleneck of deploying and running a large-scale
> job
> > in Flink is mainly focused on the following procedures:
> >
> > Procedure       Time complexity
> > Initializing ExecutionGraph
> > O(N^2)
> > Building DefaultExecutionTopology
> > O(N^2)
> > Initializing PipelinedRegionSchedulingStrategy
> > O(N^2)
> > Scheduling downstream tasks when a task finishes
> > O(N^2)
> > Calculating tasks to restart when a failover occurs
> > O(N^2)
> > Releasing result partitions
> > O(N^2)
> >
> > These procedures are all related to the complexity of the topology in the
> > ExecutionGraph. Between two vertices connected with the all-to-all edges,
> > all the upstream Intermediate ResultPartitions are connected to all
> > downstream ExecutionVertices. The computation complexity of building and
> > traversing all these edges will be O(N^2).
> >
> > As for memory usage, currently we use ExecutionEdges to store the
> > information of connections. For the all-to-all distribution type, there
> are
> > O(N^2) ExecutionEdges. We test a simple job with only two vertices. The
> > parallelisms of them are both 10k. Furthermore, they are connected with
> > all-to-all edges. It takes 4.175 GiB (estimated via MXBean) to store the
> > 100M ExecutionEdges.
> >
> > In most large-scale jobs, there will be more than two vertices with large
> > parallelisms, and they would cost a lot of time and memory to deploy the
> > job.
> >
> > As we can see, for two JobVertices connected with the all-to-all
> > distribution type, all IntermediateResultPartitions produced by the
> > upstream ExecutionVertices are isomorphic, which means that the
> downstream
> > ExecutionVertices they connected are exactly the same. The downstream
> > ExecutionVertices belonging to the same JobVertex are also isomorphic, as
> > the upstream ResultPartitions they connect are the same, too.
> >
> > Since every JobEdge has exactly one distribution type, we can divide the
> > vertices and result partitions into groups according to the distribution
> > type of the JobEdge.
> >
> > For the all-to-all distribution type, since all downstream vertices are
> > isomorphic, they belong to a single group, and all the upstream result
> > partitions are connected to this group. Vice versa, all the upstream
> result
> > partitions also belong to a single group, and all the downstream vertices
> > are connected to this group. In the past, when we wanted to iterate all
> the
> > downstream vertices, we needed to loop over them n times, which leads to
> > the complexity of O(N^2). Now since all upstream result partitions are
> > connected to one downstream group, we just need to loop over them once,
> > with the complexity of O(N).
> >
> > For the pointwise distribution type, because each result partition is
> > connected to different downstream vertices, they should belong to
> different
> > groups. Vice versa, all the vertices belong to different groups. Since
> one
> > result partition group is connected to one vertex group pointwisely, the
> > computation complexity of looping over them is still O(N).
> >
> > After we group the result partitions and vertices, ExecutionEdge is no
> > longer needed. For the test job we mentioned above, the optimization can
> > effectively reduce the memory usage from 4.175 GiB to 12.076 MiB
> (estimated
> > via MXBean) in our POC. The time cost is reduced from 62.090 seconds to
> > 8.551 seconds (with 10k parallelism).
> >
> > The detailed design doc with illustrations is located at [3]. Please find
> > more details in the links below.
> >
> > Looking forward to your feedback.
> >
> > [1] https://issues.apache.org/jira/browse/FLINK-21110
> > [2] https://issues.apache.org/jira/browse/FLINK-20612
> > [3]
> >
> https://docs.google.com/document/d/1OjGAyJ9Z6KsxcMtBHr6vbbrwP9xye7CdCtrLvf8dFYw/edit?usp=sharing
> >
> >
>