[Improvement] Optimize data flushing and memory usage for huge partitions to improve stability

[zuston]

Code of Conduct

• I agree to follow this project's Code of Conduct

Search before asking

• I have searched in the issues and found no similar issues.

What would you like to be improved?

Background

In our internal uniffle cluster, when using the storage type of MEMORY_LOCALFILE, the partial single huge partitions make disk reach high watermark, which make whole shuffle-server out-of-service due to full memory occupation.

Possible solutions

1. Using the storage type of MEMORY_LOCALFILE_HDFS with the fallback strategy of LocalStorageManagerFallbackStrategy. [ISSUE-163][FEATURE] Write to hdfs when local disk can't be write #235

After testing this mechanism, I found some bugs which make fallback invalid.

2. Introduce the new strategy of dynamic disk selection [Improvement] Optimize local disk selection strategy #373
3. Introduce partition size based strategy to flush single huge partition data to HDFS

Currently, I prefer 3th solution. In which, we could use cold storage(HDFS) when cumulative size of a particular partition is above a specific threshold(like 50g?). Actually, if exclude the partial huge partitions, the disk free ratio of whole shuffle-server is 20%-30%.

Reasons

1. We could avoid hotspot of a single shuffle server, which could use HDFS to distribute pressure.
2. Especially useful when disk spaces of shuffle servers are limited, which cannot hold a super large shuffle partition.
3. Compared with 1th solution, it is more restrained for HDFS use and will not cause greater pressure. (Actually we don't want to accept much shuffle-data to HDFS with 3 replicas)
4. Compared with 2th solution, it is more effective and have better performance.
5. Only downgrade storage of partial huge partition jobs to HDFS, which is effective, especially when the HDFS cluster performance is not good.

Final solution

The solution of handling huge partitions is to make it flush to HDFS directly and limit memory usage, all subtasks are as follows.

1. Speed up flushing partition data to HDFS.
   • Support writing multi files of single partition to improve speed in HDFS storage #396
2. Introduce the memory usage limitation for huge partitions
   • [ISSUE-378][HugePartition][Part-1] Record every partition data size for one app #458
   • [ISSUE-378][HugePartition][Part-2] Introduce memory usage limit and data flush #471
   • [ISSUE-378][HugePartition][Part-3] Introduce more metrics about huge partition #494
   • [ISSUE-378][HugePartition][Part-4] Supplement doc about huge partitions #505

How should we improve?

No response

Are you willing to submit PR?

• Yes I am willing to submit a PR!

[zuston]

After discussing with my mentor, I raise a new proposal. Feel free to discuss more. @xianjingfeng @jerqi

[jerqi]

After discussing with my mentor, I raise a new proposal. Feel free to discuss more. @xianjingfeng @jerqi

What's the advantage of this strategy compared with current strategy?

[zuston]

After discussing with my mentor, I raise a new proposal. Feel free to discuss more. @xianjingfeng @jerqi

What's the advantage of this strategy compared with current strategy?

Reasons have been attached.

[jerqi]

My concern about this strategy is that we need to maintain more partition meta. Will it give more gc pressure?

[jerqi]

After discussing with my mentor, I raise a new proposal. Feel free to discuss more. @xianjingfeng @jerqi

What's the advantage of this strategy compared with current strategy?

Reasons have been attached.

Current strategy have picked the most partition to HDFS. You can see our implement.

[zuston]

Yes, I have tested this implementation. But it looks invalid when fallback to HDFS. But this is not a big problem.

The key of proposal is to avoid too much data into HDFS, we don't want to accept shuffle data 3 replica in HDFS. And sometimes, the performance is not good.

[jerqi]

Yes, I have tested this implementation. But it looks invalid when fallback to HDFS. But this is not a big problem.

The key of proposal is to avoid too much data into HDFS, we don't want to accept shuffle data 3 replica in HDFS. And sometimes, the performance is not good.

You can set the data replica with hadoop configuration.

[zuston]

This will bring a great burden to the HDFS ops

[jerqi]

This will bring a great burden to the HDFS ops

You misunderstand me. The replica num is a client configuration. We just modify the Spark's client. You can see https://github.com/apache/incubator-uniffle/blob/884921b6d7c5452f6f3548ebc87840071c6be81f/proto/src/main/proto/Rss.proto#L422 for details.

[zuston]

Let me take a look.

[jerqi]

The data which was written to HDFS must be higher than a data size. If we write small data size, it will influence the performance of HDFS.

[zuston]

The data which was written to HDFS must be higher than a data size. If we write small data size, it will influence the performance of HDFS.

I know this, so I hope to use hdfs as little as possible, only open for those huge partition data.

[jerqi]

If you write the data which partition size > 50G && event > 64MB, I feel that it's ok.

[xianjingfeng]

In our production environment the disk is enough in most of the time. And it is difficult to set the threshold of partition size, and this will result in reduced disk usage. So we prefer to write hdfs when the disk is full.

[ChengbingLiu]

@jerqi Here is our scenario:

We co-located Uniffle shuffle servers with our YARN NodeManagers. Therefore the shuffle servers have limited memory (10-20 GB) and limited disk space (1-5 TB). However, there can be as many as 1k shuffle tasks on a shuffle server at the same time, with each task takes up only 1~2MB memory on the average.

When a super large shuffle partition comes in, one of the disks quickly reaches its high watermark and memory starts to flush directly to HDFS. Since each task has only a small size (1-2MB) in the memory, it could flush(append) to HDFS with only 1~2MB at a time even if the configured flush threshold rss.server.flush.cold.storage.threshold.size is 64MB.

IMO, the flush threshold works only with two preconditions:

1. The shuffle server has enough memory compared to concurrent shuffle tasks (otherwise the threshold may not be reached)
2. The shuffle server has enough disk space (otherwise small partitions may fallback to cold storage)

If either condition is not met, it's better to have another machanism to better utilize the limited memory/disk resources, instead of wasting them on a uncommonly large partition.

Please correct me if I'm wrong.

[jerqi]

@jerqi Here is our scenario:

We co-located Uniffle shuffle servers with our YARN NodeManagers. Therefore the shuffle servers have limited memory (10-20 GB) and limited disk space (1-5 TB). However, there can be as many as 1k shuffle tasks on a shuffle server at the same time, with each task takes up only 1~2MB memory on the average.

When a super large shuffle partition comes in, one of the disks quickly reaches its high watermark and memory starts to flush directly to HDFS. Since each task has only a small size (1-2MB) in the memory, it could flush(append) to HDFS with only 1~2MB at a time even if the configured flush threshold rss.server.flush.cold.storage.threshold.size is 64MB.

IMO, the flush threshold works only with two preconditions:

1. The shuffle server has enough memory compared to concurrent shuffle tasks (otherwise the threshold may not be reached)
2. The shuffle server has enough disk space (otherwise small partitions may fallback to cold storage)

If either condition is not met, it's better to have another machanism to better utilize the limited memory/disk resources, instead of wasting them on a uncommonly large partition.

Please correct me if I'm wrong.

Yes, you're right. But if we only write 2MB to HDFS directly, the HDFS will have poor performance.

[zuston]

In our production environment the disk is enough in most of the time. And it is difficult to set the threshold of partition size, and this will result in reduced disk usage. So we prefer to write hdfs when the disk is full.

Yes, this is difficult to set the threshold of partition size. I still think existing mechanism is hard to solve the huge single partition, especially in our scenario mentioned by @ChengbingLiu . I want to share some practices that wants to solve this problem but failed.

First case:

I enable the storage type of MEMORY_LOCALFILE_HDFS and the conf of rss.server.flush.cold.storage.threshold.size is 100G. The buffer capacity is 10GB. The storage fallback strategy is LocalStorageFallbackStrategy.

When a huge single partition came and quickly it made the disk reach the high-watermark, it started to flush HDFS directly. However, unfortunately, because this huge partition occupies a large space in memory(6GB), the shuffle dataFlushEvent is large, which leads to the need to spend a longer time to consume. And this will cause other jobs to be unable to write data normally. The picture is shown below.

Second case:

After above practice, I set the rss.server.flush.cold.storage.threshold.size to 128M, which means the partition buffer once reaches the 128M, it will directly flush to HDFS.

However, unfortunately, this still is useless. Although the huge partition will generate several dataFlushEvents waiting to be flushed to a single HDFS file, but its writerHandler will hold the write lock to write one by one. So this will cause the flush thread pool busy but the writing speed is still slow.

1. And this also will cause other jobs to be unable to write data normally.
2. Especially, when in one shuffle-server, there are not many partitions, it will cause the local disk always empty. This is unreasonable.

The case's picture is shown below.

Possible solution

The above problem is hard to avoid in production environment. I think we need to introduce new strategy to solve. The following solutions are just draft

1. Introduce the blacklist of a app based on its flushed size, once reached, shuffle-server will let it allow to get limited memory and flush data to HDFS directly
2. Client could get the multiple candidates(multiple shuffle-servers for one partition, not multiple replica). Once written failed in one shuffle-server, it could write data to another shuffle-server, which will distribute the huge data pressure to uniffle cluster. After all, the overall disk utilization rate of the cluster is relatively low

[jerqi]

Maybe we could introduce multi-thread writing HDFS. If the file is too big, we could split them to multiple files. ByteDance CSS have similar concept. If file exceed the size, we will open and write another file.

[zuston]

Maybe we could introduce multi-thread writing HDFS. If the file is too big, we could split them to multiple files.

Yes. The key of problem is the low speed of writing single one data file.

ByteDance CSS have similar concept. If file exceed the size, we will open and write another file.

Let me take a look. But I think writing another file is not a good solution, which wont improve the writing concurrency for multiple same partition events.

[jerqi]

Maybe we could introduce multi-thread writing HDFS. If the file is too big, we could split them to multiple files.

Yes. The key of problem is the low speed of writing single one data file.

ByteDance CSS have similar concept. If file exceed the size, we will open and write another file.

Let me take a look. But I think writing another file is not a good solution, which wont improve the writing concurrency for multiple same partition events.

I mean that we can write multiple files at the same time.
We can tryLock the file lock, we have multiple locks, if we succeed to tryLock, we can write the file. If we fail to tryLock, we will retry another file lock.

[zuston]

I mean that we can write multiple files at the same time.
We can tryLock the file lock, we have multiple locks, if we succeed to tryLock, we can write the file. If we fail to tryLock, we will retry another file lock.

Got it. It's an optional solution, the flush speed could improve based on (HDFS speed) * (file number).

And I think we also need to limit the huge partition app's writing speed to avoid affecting other jobs.

[jerqi]

I mean that we can write multiple files at the same time.
We can tryLock the file lock, we have multiple locks, if we succeed to tryLock, we can write the file. If we fail to tryLock, we will retry another file lock.

Got it. It's an optional solution, the flush speed could improve based on (HDFS speed) * (file number).

And I think we also need to limit the huge partition app's writing speed to avoid affecting other jobs.

VIP Shop have similarly minds. cc @Gustfh

[jerqi]

I mean that we can write multiple files at the same time.
We can tryLock the file lock, we have multiple locks, if we succeed to tryLock, we can write the file. If we fail to tryLock, we will retry another file lock.

Got it. It's an optional solution, the flush speed could improve based on (HDFS speed) * (file number).

I feel that it's necessary. Because it will be too slow that the server only use single thread to flush a large partition although we have multiple servers.

[zuston]

I feel that it's necessary. Because it will be too slow that the server only use single thread to flush a large partition although we have multiple servers.

Make sense. I think I will do this optimization.

[zuston]

I feel that it's necessary. Because it will be too slow that the server only use single thread to flush a large partition although we have multiple servers.

Make sense. I think I will do this optimization.

Draft a PR #396 .

For this PR, it only solve the problem mentioned by second case. But if the ShuffleDataFlushEvent is so huge(5G) to HDFS, it's still handled by single one thread. So we also need to generate multiple events from huge ShuffleDataFlushEvent to improve writing speed when flushing to HDFS.

[zuston]

Introduce the blacklist of a app based on its flushed size, once reached, shuffle-server will let it allow to get limited memory and flush data to HDFS directly

If the huge partition has the high writing speed and the speed of flushing is slow, the most memory will be occupied by huge partition's app. For ensuring the other apps stability, I prefer limiting the memory usage for huge partition. WDYT? @jerqi cc @advancedxy

[zuston]

This issue will track all the optimizations of huge partition, and all sub-tasks will be connected with this. The solution of handling huge partitions is to make it flush to HDFS directly and limit memory usage, all subtasks are as follows.

1. Speed up flushing partition data to HDFS.
   • Support writing multi files of single partition to improve speed in HDFS storage #396
2. Introduce the memory usage limitation for huge partitions
   • [ISSUE-378][HugePartition][Part-1] Record every partition data size for one app #458
   • [ISSUE-378][HugePartition][Part-2] Introduce memory usage limit and data flush #471
   • [ISSUE-378][HugePartition][Part-3] Introduce more metrics about huge partition #494
   • [ISSUE-378][HugePartition][Part-4] Supplement doc about huge partitions #505
   • Introduce storage selector strategy(to support huge partition flushed to HDFS directly) in MultipleStorageManager to replace fallback strategy
3. Support split huge event into smaller multiple events concurrently to speed up flushing

cc @jerqi @advancedxy I will create some subtasks of issues and PRs, feel free to discuss more.

[jerqi]

This issue will track all the optimizations of huge partition, and all sub-tasks will be connected with this. The solution of handling huge partitions is to make it flush to HDFS directly and limit memory usage, all subtasks are as follows.

1. Speed up flushing partition data to HDFS.

   • Support writing multi files of single partition to improve speed in HDFS storage #396

2. Introduce the memory usage limitation for huge partitions

   • Record every partition data size in ShuffleTaskInfo
   • Introduce storage selector strategy(to support huge partition flushed to HDFS directly) in MultipleStorageManager to replace fallback strategy
   • Introduce more metrics to monitor huge partitions and so on

3. Support split huge event into smaller multiple events concurrently to speed up flushing

cc @jerqi @advancedxy I will create some subtasks of issues and PRs, feel free to discuss more.

It's ok for me.

[advancedxy]

This issue will track all the optimizations of huge partition, and all sub-tasks will be connected with this. The solution of handling huge partitions is to make it flush to HDFS directly and limit memory usage, all subtasks are as follows.

1. Speed up flushing partition data to HDFS.

   • Support writing multi files of single partition to improve speed in HDFS storage #396

2. Introduce the memory usage limitation for huge partitions

   • [ISSUE-378][HugePartition][Part-1] Record every partition data size for one app #458
   • Introduce storage selector strategy(to support huge partition flushed to HDFS directly) in MultipleStorageManager to replace fallback strategy
   • Introduce more metrics to monitor huge partitions and so on

3. Support split huge event into smaller multiple events concurrently to speed up flushing

cc @jerqi @advancedxy I will create some subtasks of issues and PRs, feel free to discuss more.

Missed this comment. LGTM.

[zuston]

Closed. Feel free to open it if having any problem.